You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "letaoj (via GitHub)" <gi...@apache.org> on 2025/10/01 22:55:19 UTC
[PR] [example][java] Add ReAct agent example [flink-agents]
letaoj opened a new pull request, #242:
URL: https://github.com/apache/flink-agents/pull/242
<!--
* Thank you very much for contributing to Flink Agents.
* Please add the relevant components in the PR title. E.g., [api], [runtime], [java], [python], [hotfix], etc.
-->
<!-- Please link the PR to the relevant issue(s). Hotfix doesn't need this. -->
Linked issue: #xxx
### Purpose of change
<!-- What is the purpose of this change? -->
### Tests
<!-- How is this change verified? -->
```
❯ cd examples; java --add-opens java.base/java.util=ALL-UNNAMED --add-opens java.base/java.lang=ALL-UNNAMED --add-opens java.base/java.lang.reflect=ALL-UNNAMED --add-opens java.base/java.text=ALL-UNNAMED --add-opens java.desktop/java.awt.font=ALL-UNNAMED -cp "target/classes:$(mvn dependency:build-classpath -q -Dmdep.outputFile=/dev/stdout)" org.apache.flink.agents.examples.ReActAgentExample; cd ..
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by com.esotericsoftware.kryo.util.UnsafeUtil (file:/Users/ljiang/.m2/repository/com/esotericsoftware/kryo/kryo/2.24.0/kryo-2.24.0.jar) to constructor java.nio.DirectByteBuffer(long,int,java.lang.Object)
WARNING: Please consider reporting this to the maintainers of com.esotericsoftware.kryo.util.UnsafeUtil
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
8> ProductReviewAnalysisRes{id='B009MA34NY', score=5, reasons=[good support, comfortable, lightweight, well-made, long lasting]}
4> ProductReviewAnalysisRes{id='B009MA34NY', score=5, reasons=[]}
1> ProductReviewAnalysisRes{id='B000YFSR5G', score=2, reasons=[did not fit well, not comfortable]}
5> ProductReviewAnalysisRes{id='B009MA34NY', score=5, reasons=[]}
7> ProductReviewAnalysisRes{id='B001IKJOLW', score=5, reasons=[]}
9> ProductReviewAnalysisRes{id='B000YFSR4W', score=5, reasons=[]}
6> ProductReviewAnalysisRes{id='B010RRWKT4', score=5, reasons=[]}
11> ProductReviewAnalysisRes{id='B000YFSR5G', score=5, reasons=[]}
10> ProductReviewAnalysisRes{id='B010RRWKT4', score=5, reasons=[comfortable]}
12> ProductReviewAnalysisRes{id='B014IBJKNO', score=5, reasons=[]}
3> ProductReviewAnalysisRes{id='B0092UF54A', score=5, reasons=[excellent product quality]}
8> ProductReviewAnalysisRes{id='B010RRWKT4', score=5, reasons=[]}
2> ProductReviewAnalysisRes{id='B000YFSR4W', score=1, reasons=[damaged product upon delivery]}
4> ProductReviewAnalysisRes{id='B000KPIHQ4', score=5, reasons=[]}
1> ProductReviewAnalysisRes{id='B0058YEJ5K', score=5, reasons=[comfortable]}
5> ProductReviewAnalysisRes{id='B005AGO4LU', score=5, reasons=[comfortable, great for weight training]}
7> ProductReviewAnalysisRes{id='B0058YEJ5K', score=2, reasons=[poor quality, overpriced]}
```
### API
<!-- Does this change touches any public APIs? -->
### Documentation
<!-- Should this change be covered by the user documentation?-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [example][java] Add ReAct agent example [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #242:
URL: https://github.com/apache/flink-agents/pull/242#discussion_r2403896328
##########
examples/src/main/java/org/apache/flink/agents/examples/ReActAgentExample.java:
##########
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.examples;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.AgentsExecutionEnvironment;
+import org.apache.flink.agents.api.agents.ReActAgent;
+import org.apache.flink.agents.api.annotation.Prompt;
+import org.apache.flink.agents.api.annotation.Tool;
+import org.apache.flink.agents.api.annotation.ToolParam;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.examples.agents.CustomTypesAndResources;
+import org.apache.flink.agents.integrations.chatmodels.ollama.OllamaChatModelSetup;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.connector.file.src.FileSource;
+import org.apache.flink.connector.file.src.reader.TextLineFormat;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.types.Row;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Objects;
+
+import static org.apache.flink.agents.api.resource.ResourceType.TOOL;
+
+/**
+ * Java example demonstrating the ReActAgent for product review analysis and shipping notification.
+ *
+ * <p>This example demonstrates how to use the Flink Agents to analyze product reviews and record
+ * shipping questions in a streaming pipeline. The pipeline reads product reviews from a source,
+ * deserializes each review, and uses an LLM agent to extract review scores and unsatisfied reasons.
+ * If the unsatisfied reasons are related to shipping, the agent will notify the shipping manager.
+ * This serves as a minimal, end-to-end example of integrating LLM-powered react agent with Flink
+ * streaming jobs.
+ */
+public class ReActAgentExample {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ /**
+ * Tool for notifying the shipping manager when product received a negative review due to
+ * shipping damage.
+ *
+ * @param id The id of the product that received a negative review due to shipping damage
+ * @param review The negative review content
+ */
+ @Tool(
+ description =
+ "Notify the shipping manager when product received a negative review due to shipping damage.")
+ public static void notifyShippingManager(
+ @ToolParam(name = "id") String id, @ToolParam(name = "review") String review) {
+ notifyShippingManager(id, review);
Review Comment:
Should use `org.apache.flink.agents.examples.agents.CustomTypesAndResources.notifyShippingManager(
id, review);` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [example][java] Add ReAct agent example [flink-agents]
Posted by "letaoj (via GitHub)" <gi...@apache.org>.
letaoj commented on PR #242:
URL: https://github.com/apache/flink-agents/pull/242#issuecomment-3358446980
@wenjin272 @xintongsong @Sxnan please take a look when you got time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [example][java] Add ReAct agent example [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #242:
URL: https://github.com/apache/flink-agents/pull/242#discussion_r2405842245
##########
examples/src/main/java/org/apache/flink/agents/examples/ReActAgentExample.java:
##########
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.examples;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.AgentsExecutionEnvironment;
+import org.apache.flink.agents.api.agents.ReActAgent;
+import org.apache.flink.agents.api.annotation.Prompt;
+import org.apache.flink.agents.api.annotation.Tool;
+import org.apache.flink.agents.api.annotation.ToolParam;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.examples.agents.CustomTypesAndResources;
+import org.apache.flink.agents.integrations.chatmodels.ollama.OllamaChatModelSetup;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.connector.file.src.FileSource;
+import org.apache.flink.connector.file.src.reader.TextLineFormat;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.types.Row;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Objects;
+
+import static org.apache.flink.agents.api.resource.ResourceType.TOOL;
+
+/**
+ * Java example demonstrating the ReActAgent for product review analysis and shipping notification.
+ *
+ * <p>This example demonstrates how to use the Flink Agents to analyze product reviews and record
+ * shipping questions in a streaming pipeline. The pipeline reads product reviews from a source,
+ * deserializes each review, and uses an LLM agent to extract review scores and unsatisfied reasons.
+ * If the unsatisfied reasons are related to shipping, the agent will notify the shipping manager.
+ * This serves as a minimal, end-to-end example of integrating LLM-powered react agent with Flink
+ * streaming jobs.
+ */
+public class ReActAgentExample {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ /**
+ * Tool for notifying the shipping manager when product received a negative review due to
+ * shipping damage.
+ *
+ * @param id The id of the product that received a negative review due to shipping damage
+ * @param review The negative review content
+ */
+ @Tool(
+ description =
+ "Notify the shipping manager when product received a negative review due to shipping damage.")
+ public static void notifyShippingManager(
+ @ToolParam(name = "id") String id, @ToolParam(name = "review") String review) {
+ notifyShippingManager(id, review);
Review Comment:
Maybe you can add @Tool and @ToolParam directly on the `notifyShippingManager ` method in CustomTypesAndResources, and register it in agents execution envrionment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [example][java] Add ReAct agent example [flink-agents]
Posted by "letaoj (via GitHub)" <gi...@apache.org>.
letaoj commented on code in PR #242:
URL: https://github.com/apache/flink-agents/pull/242#discussion_r2405055476
##########
examples/src/main/java/org/apache/flink/agents/examples/ReActAgentExample.java:
##########
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.examples;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.AgentsExecutionEnvironment;
+import org.apache.flink.agents.api.agents.ReActAgent;
+import org.apache.flink.agents.api.annotation.Prompt;
+import org.apache.flink.agents.api.annotation.Tool;
+import org.apache.flink.agents.api.annotation.ToolParam;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.examples.agents.CustomTypesAndResources;
+import org.apache.flink.agents.integrations.chatmodels.ollama.OllamaChatModelSetup;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.connector.file.src.FileSource;
+import org.apache.flink.connector.file.src.reader.TextLineFormat;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.types.Row;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Objects;
+
+import static org.apache.flink.agents.api.resource.ResourceType.TOOL;
+
+/**
+ * Java example demonstrating the ReActAgent for product review analysis and shipping notification.
+ *
+ * <p>This example demonstrates how to use the Flink Agents to analyze product reviews and record
+ * shipping questions in a streaming pipeline. The pipeline reads product reviews from a source,
+ * deserializes each review, and uses an LLM agent to extract review scores and unsatisfied reasons.
+ * If the unsatisfied reasons are related to shipping, the agent will notify the shipping manager.
+ * This serves as a minimal, end-to-end example of integrating LLM-powered react agent with Flink
+ * streaming jobs.
+ */
+public class ReActAgentExample {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ /**
+ * Tool for notifying the shipping manager when product received a negative review due to
+ * shipping damage.
+ *
+ * @param id The id of the product that received a negative review due to shipping damage
+ * @param review The negative review content
+ */
+ @Tool(
+ description =
+ "Notify the shipping manager when product received a negative review due to shipping damage.")
+ public static void notifyShippingManager(
+ @ToolParam(name = "id") String id, @ToolParam(name = "review") String review) {
+ notifyShippingManager(id, review);
Review Comment:
Thanks for catching this. This is not an obvious bug. Is that possible to write this code in a different way so that I do not need to have this sort of redundent call in the ReAct agent?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [example][java] Add ReAct agent example [flink-agents]
Posted by "letaoj (via GitHub)" <gi...@apache.org>.
letaoj commented on PR #242:
URL: https://github.com/apache/flink-agents/pull/242#issuecomment-3358445635
One thing that I'm not sure is whether the notifyShipping tools was called or not. At least, i'm not able to see it from the short period that I ran it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [example][java] Add ReAct agent example [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #242:
URL: https://github.com/apache/flink-agents/pull/242#issuecomment-3375451167
LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [example][java] Add ReAct agent example [flink-agents]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #242:
URL: https://github.com/apache/flink-agents/pull/242#discussion_r2397751701
##########
examples/src/main/java/org/apache/flink/agents/examples/ReActAgentExample.java:
##########
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.examples;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.AgentsExecutionEnvironment;
+import org.apache.flink.agents.api.agents.ReActAgent;
+import org.apache.flink.agents.api.annotation.Prompt;
+import org.apache.flink.agents.api.annotation.Tool;
+import org.apache.flink.agents.api.annotation.ToolParam;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.examples.agents.CustomTypesAndResources;
+import org.apache.flink.agents.integrations.chatmodels.ollama.OllamaChatModelSetup;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.connector.file.src.FileSource;
+import org.apache.flink.connector.file.src.reader.TextLineFormat;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.types.Row;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Objects;
+
+import static org.apache.flink.agents.api.resource.ResourceType.TOOL;
+
+/**
+ * Java example demonstrating the ReActAgent for product review analysis and shipping notification.
+ *
+ * <p>This example demonstrates how to use the Flink Agents to analyze product reviews and record
+ * shipping questions in a streaming pipeline. The pipeline reads product reviews from a source,
+ * deserializes each review, and uses an LLM agent to extract review scores and unsatisfied reasons.
+ * If the unsatisfied reasons are related to shipping, the agent will notify the shipping manager.
+ * This serves as a minimal, end-to-end example of integrating LLM-powered react agent with Flink
+ * streaming jobs.
+ */
+public class ReActAgentExample {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ /**
+ * Tool for notifying the shipping manager when product received a negative review due to
+ * shipping damage.
+ *
+ * @param id The id of the product that received a negative review due to shipping damage
+ * @param review The negative review content
+ */
+ @Tool(
+ description =
+ "Notify the shipping manager when product received a negative review due to shipping damage.")
+ public static void notifyShippingManager(
+ @ToolParam(name = "id") String id, @ToolParam(name = "review") String review) {
+ notifyShippingManager(id, review);
+ }
+
+ @Prompt
+ public static org.apache.flink.agents.api.prompt.Prompt reviewAnalysisReactPrompt() {
+ return CustomTypesAndResources.REVIEW_ANALYSIS_REACT_PROMPT;
+ }
+
+ /** Runs the example pipeline. */
+ public static void main(String[] args) throws Exception {
+ // Set up the Flink streaming environment and the Agents execution environment.
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ AgentsExecutionEnvironment agentsEnv =
+ AgentsExecutionEnvironment.getExecutionEnvironment(env);
+
+ // Add Ollama chat model connection and record shipping question tool to be used
+ // by the Agent.
+ agentsEnv
+ .addResource(
+ "ollamaChatModelConnection",
+ ResourceType.CHAT_MODEL_CONNECTION,
+ CustomTypesAndResources.OLLAMA_SERVER_DESCRIPTOR)
+ .addResource(
+ "notifyShippingManager",
+ TOOL,
+ org.apache.flink.agents.api.tools.Tool.fromMethod(
+ ReActAgentExample.class.getMethod(
+ "notifyShippingManager", String.class, String.class)));
+
+ // Read product reviews from input_data.txt file as a streaming source.
+ // Each element represents a ProductReview.
+ DataStream<Row> productReviewStream =
+ env.fromSource(
+ FileSource.forRecordStreamFormat(
+ new TextLineFormat(),
Review Comment:
`TextLineFormat` has been deprecated and removed in Flink 2.x. I think we should use `TextLineInputFormat` instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [example][java] Add ReAct agent example [flink-agents]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong merged PR #242:
URL: https://github.com/apache/flink-agents/pull/242
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org