wenjin272 opened a new issue, #1080: URL: https://github.com/apache/flink-agents/issues/1080
### Search before asking - [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description This is a child issue of #1055. #### Motivation An Agent can currently emit arbitrary values through `OutputEvent.output`. This flexibility should be preserved. However, the DataStream/Table integration currently exposes the internal opaque output representation directly and requires users to reconstruct type information at the terminal API boundary. In Java: - `AgentBuilder.toDataStream()` returns `DataStream<Object>`. - The Agent operator is created with `TypeInformation.of(Object.class)`. - `toTable(Schema)` applies a Table schema over that untyped stream. In Python: - The Java/Python bridge produces `DataStream<byte[]>` and the Python side unpickles each value. - `to_datastream()` uses the default pickle output type unless a `TypeInformation` is supplied to the implementation-specific method. - `to_table()` requires both a `Schema` and a matching `TypeInformation`, so users describe the same physical fields twice in common cases. - The public abstract `to_datastream()` signature does not expose the optional `output_type` accepted by the remote implementation. - The deserialized stream is cached after the first terminal call. As a result, calling untyped `to_datastream()` before typed `to_table()` can cause the later `output_type` to be ignored. The integration tests also leak the target representation into Agent logic: a Python Agent that produces a Pydantic model for DataStream output has to manually convert the same logical result to `Row` for Table output. Related issues #772 and #777 documented the existing Java/Python API and documentation differences, but did not address the underlying output typing and materialization model. Flink requires an operator's output `TypeInformation` while constructing the job graph. It cannot infer a reliable type from future runtime records when an Agent is allowed to emit arbitrary or heterogeneous values. Therefore, raw Agent execution and typed Flink output should be modeled separately. #### Proposed direction Separate the unrestricted Agent execution result from an optional typed output view. For example (names are illustrative): ```python execution = agents_env.from_datastream(input_stream, key_selector).apply(agent) # Preserve unrestricted Agent output semantics. raw_stream = execution.to_datastream() # Create one typed view when DataStream/Table type semantics are needed. typed_output = execution.output_as(ReviewOutput) typed_stream = typed_output.to_datastream() table = typed_output.to_table() ``` ```java AgentExecution execution = agentsEnv.fromDataStream(inputStream, keySelector).apply(agent); DataStream<Object> rawStream = execution.toDataStream(); AgentOutput<ReviewOutput> typedOutput = execution.outputAs(TypeInformation.of(ReviewOutput.class)); DataStream<ReviewOutput> typedStream = typedOutput.toDataStream(); Table table = typedOutput.toTable(); ``` The typed output descriptor should be specified or inferred once and provide all information needed for: - the DataStream runtime `TypeInformation`; - conversion and validation of emitted values; - deriving the physical Table schema; - adapting Python models or dictionaries to a Table-compatible `Row` representation. Python inference should investigate common structured types such as `TypedDict`, dataclasses, named tuples, and Pydantic models, following the same general principle as PyFlink's type-hint-based UDF/DataFrame inference. An explicit descriptor remains the fallback when inference is unavailable. `Schema` should remain available for Table-specific enrichment such as renaming, computed columns, metadata columns, or watermarks, rather than being a second mandatory declaration of the physical output type. The raw Agent operator may continue to use `Object` or serialized bytes internally. A typed view can either provide the Agent operator's `TypeInformation` before construction where possible, or add a downstream conversion operator with a known output type. Requesting a typed view requires its records to conform to one homogeneous output contract; unrestricted heterogeneous output remains available through the raw view. Compatibility and migration mechanics are outside the scope of this issue. #### Acceptance criteria - Agents can still emit arbitrary values through a clearly defined raw output mode. - A typed DataStream/Table view uses one output type declaration or inference result instead of separate matching `Schema` and `TypeInformation` declarations. - Normal Python `to_table()` usage no longer requires users to construct both `ExternalTypeInfo(RowTypeInfo(...))` and an equivalent physical `Schema`. - Table-specific `Schema` enrichment remains supported as an optional operation. - Python Agents do not need to manually change a logical output from a Pydantic model to `Row` solely because the caller chooses Table output. - Repeated or differently ordered calls to DataStream/Table materialization APIs do not change or silently ignore output type information. - Java and Python expose equivalent raw-versus-typed output semantics and clear failure behavior when values do not match a requested typed view. - Integration tests cover DataStream-to-DataStream, DataStream-to-Table, and Table-to-Table paths in both Java and Python. - Documentation explains when output typing can be inferred, when an explicit descriptor is required, and how heterogeneous output is handled. ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- 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]
