wenjin272 commented on code in PR #1312:
URL: https://github.com/apache/flink-agents/pull/1312#discussion_r4355491346
##########
examples/src/main/java/org/apache/flink/agents/examples/WorkflowSingleAgentExample.java:
##########
@@ -80,7 +80,7 @@ public static void main(String[] args) throws Exception {
// Use the ReviewAnalysisAgent to analyze each product review.
DataStream<Object> reviewAnalysisResStream =
agentsEnv
- .fromDataStream(productReviewStream)
+ .fromDataStream(productReviewStream, value -> value)
Review Comment:
Could we also align the Java examples with the Python examples by using
typed records and product IDs as keys, instead of adding `value -> value`
everywhere?
For the single-agent and YAML examples, deserialize JSON into
`ProductReview` upstream:
```java
DataStream<ProductReview> productReviewStream =
rawStream
.map(json -> MAPPER.readValue(json, ProductReview.class))
.returns(ProductReview.class);
DataStream<Object> output =
agentsEnv
.fromDataStream(productReviewStream, ProductReview::getId)
.apply(new ReviewAnalysisAgent())
.toDataStream();
```
The YAML example can use the same typed input and selector while keeping
`.apply("review_analysis_agent")`. `ReviewAnalysisAgent.processInput` would
consume the object directly:
```java
ProductReview review =
(ProductReview) InputEvent.fromEvent(event).getInput();
ctx.getShortTermMemory().set("id", review.getId());
// Build the prompt from review.getId() and review.getReview().
```
This keeps JSON parsing at the input boundary and avoids parsing again
inside the selector or agent.
For the multi-agent example, let the window emit `ProductReviewSummary`
directly:
```java
// Change the window function's output type to ProductReviewSummary:
// ProcessWindowFunction<
// ProductReviewAnalysisRes, ProductReviewSummary, String,
TimeWindow>
ProductReviewSummary summary =
new ProductReviewSummary(key, formattedPercentages, reasonList);
out.collect(summary); // Instead of MAPPER.writeValueAsString(summary).
```
Then key the typed stream by product ID:
```java
DataStream<ProductReviewSummary> summaries =
reviewAnalysisResStream
.keyBy(ProductReviewAnalysisRes::getId)
.window(TumblingProcessingTimeWindows.of(Duration.ofMinutes(1)))
.process(new AggregateScoreDistributionAndDislikeReasons());
DataStream<Object> suggestions =
agentsEnv
.fromDataStream(summaries, ProductReviewSummary::getId)
.apply(new ProductSuggestionAgent())
.toDataStream();
```
`ProductSuggestionAgent` can consume the object and serialize it when
constructing the prompt:
```java
ProductReviewSummary summary =
(ProductReviewSummary) InputEvent.fromEvent(event).getInput();
String promptInput = MAPPER.writeValueAsString(summary);
```
The ReAct example already produces `Row` values, so it only needs:
```java
.fromDataStream(productReviewStream, row -> (String) row.getField("id"))
```
Please update the affected callers and existing tests for these input-type
changes, and synchronize the Java snippets in the single-agent, multi-agent,
ReAct, and YAML Quickstart pages. Those snippets still use
`fromDataStream(input)` without a selector and will fail under the new
validation.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/env/RemoteExecutionEnvironment.java:
##########
@@ -229,10 +237,25 @@ public DataStream<Object> toDataStream() {
if (keySelector != null) {
outputDataStream =
CompileUtils.connectToAgent(inputDataStream,
keySelector, agentPlan);
- } else {
- // If no key selector provided, use a simple pass-through
key selector
+ } else if (inputDataStream instanceof KeyedStream) {
+ // An already-keyed input carries its business key;
connect through the
+ // keyed-stream path to preserve that key, instead of
replacing it with a
+ // pass-through selector that would split a session across
its messages.
+ outputDataStream =
+ CompileUtils.connectToAgent(
+ (KeyedStream<T, ?>) inputDataStream,
agentPlan);
+ } else if (allowIdentityKeyFallback) {
Review Comment:
I’d prefer to align the Table path too: require an explicit selector in both
Java and Python. Could we remove `allowIdentityKeyFallback`, reject Java Table
input without a selector, and update the docs? The “without keying” Javadoc is
misleading, since identity keying still determines memory sharing.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
