This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 3320e5675 [docs][python] Fix setup, Quickstart and scalar Table output
(#1316)
3320e5675 is described below
commit 3320e56752bace0574fa005e342660e0d557ab96
Author: Yu Tang <[email protected]>
AuthorDate: Thu Oct 22 14:43:25 2026 +0800
[docs][python] Fix setup, Quickstart and scalar Table output (#1316)
Generated-by: Codex CLI 0.163.0-alpha.11.1 (GPT-6)
---
.../docs/development/integrate_with_flink.md | 5 +++
.../quickstart/create-agent/multiple-agents.md | 3 ++
.../quickstart/create-agent/single-agent.md | 1 +
.../docs/get-started/quickstart/yaml_agent.md | 2 +-
docs/content/docs/get-started/setup.md | 9 ++++--
docs/layouts/partials/docs/menu-filetree.html | 13 ++++++--
docs/layouts/partials/docs/title.html | 10 +-----
examples/src/main/resources/input_data.txt | 3 +-
.../flink_integration_agent.py | 9 ++++++
.../flink_intergration_test.py | 37 ++++++++++++++++++++++
.../quickstart/workflow_single_agent_example.py | 1 +
python/flink_agents/runtime/output_type_utils.py | 21 +++++++++---
.../runtime/tests/test_output_type_utils.py | 31 ++++++++++++++++--
.../tests/test_remote_execution_environment.py | 26 +++++++++++++++
14 files changed, 148 insertions(+), 23 deletions(-)
diff --git a/docs/content/docs/development/integrate_with_flink.md
b/docs/content/docs/development/integrate_with_flink.md
index 9ec8ea98b..1bea16897 100644
--- a/docs/content/docs/development/integrate_with_flink.md
+++ b/docs/content/docs/development/integrate_with_flink.md
@@ -157,6 +157,11 @@ AgentsExecutionEnvironment agentsEnv =
Integrate the agent with input `Table`, and return the output `Table` can be
consumed by downstream.
+The input `Table` must be **insert-only**: it may append rows, but must not
update or
+delete earlier rows. Both `from_table()` and `fromTable()` reject Tables that
produce
+updates or deletes, such as a continuous `GROUP BY` aggregation whose results
change
+as new records arrive.
+
{{< tabs "From/To Table" >}}
{{< tab "Python" >}}
diff --git
a/docs/content/docs/get-started/quickstart/create-agent/multiple-agents.md
b/docs/content/docs/get-started/quickstart/create-agent/multiple-agents.md
index 49216fa83..691aab0dd 100644
--- a/docs/content/docs/get-started/quickstart/create-agent/multiple-agents.md
+++ b/docs/content/docs/get-started/quickstart/create-agent/multiple-agents.md
@@ -542,3 +542,6 @@ $FLINK_HOME/bin/flink run -c
org.apache.flink.agents.examples.WorkflowMultipleAg
The job appears in the Flink web UI at
[localhost:8081](http://localhost:8081). Suggestions
appear in the TaskManager output log under `$FLINK_HOME/log` once a window
closes, which takes
up to a minute after the first review is analyzed.
+
+This example reads a finite file, so the job may finish before a
processing-time window
+fires. Some or all suggestions may therefore not be produced.
diff --git
a/docs/content/docs/get-started/quickstart/create-agent/single-agent.md
b/docs/content/docs/get-started/quickstart/create-agent/single-agent.md
index bdacd2baa..24af388fa 100644
--- a/docs/content/docs/get-started/quickstart/create-agent/single-agent.md
+++ b/docs/content/docs/get-started/quickstart/create-agent/single-agent.md
@@ -55,6 +55,7 @@ caps how many chat requests run at once.
```python
# Set up the Flink streaming environment and the Agents execution environment.
env = StreamExecutionEnvironment.get_execution_environment()
+env.set_parallelism(1)
agents_env = AgentsExecutionEnvironment.get_execution_environment(env)
# limit async request to avoid overwhelming ollama server
diff --git a/docs/content/docs/get-started/quickstart/yaml_agent.md
b/docs/content/docs/get-started/quickstart/yaml_agent.md
index eb37be532..6f041eb92 100644
--- a/docs/content/docs/get-started/quickstart/yaml_agent.md
+++ b/docs/content/docs/get-started/quickstart/yaml_agent.md
@@ -376,5 +376,5 @@ $FLINK_HOME/bin/flink run -c
org.apache.flink.agents.examples.YamlWorkflowAgentE
{{< /tabs >}}
The job appears in the Flink web UI at
[localhost:8081](http://localhost:8081). After a few
-minutes, one JSON record per input review appears in the TaskManager output
log under
+minutes, one analysis result per input review appears in the TaskManager
output log under
`$FLINK_HOME/log`.
diff --git a/docs/content/docs/get-started/setup.md
b/docs/content/docs/get-started/setup.md
index f9eb97b7e..bf9b82cb1 100644
--- a/docs/content/docs/get-started/setup.md
+++ b/docs/content/docs/get-started/setup.md
@@ -183,8 +183,14 @@ pip install flink-agents apache-flink==${FLINK_VERSION}
```shell
echo "setuptools<82" > /tmp/constraint.txt
-PIP_CONSTRAINT=/tmp/constraint.txt pip install flink-agents
apache-flink==${FLINK_VERSION}
+PIP_CONSTRAINT=/tmp/constraint.txt \
+PIP_BUILD_CONSTRAINT=/tmp/constraint.txt \
+ pip install flink-agents apache-flink==${FLINK_VERSION}
```
+
+Set both variables for compatibility with older pip versions. Since pip 26.2,
+`PIP_CONSTRAINT` no longer affects isolated builds, which require
+[`PIP_BUILD_CONSTRAINT`](https://pip.pypa.io/en/stable/user_guide/#build-constraints).
{{< /hint >}}
#### From Source
@@ -206,7 +212,6 @@ pip install apache-flink==${FLINK_VERSION}
# Build Flink Agents
cd flink-agents
./tools/build.sh
-cd ..
```
On Apple Silicon, that `pip install` can fail while building `apache-beam`. See
diff --git a/docs/layouts/partials/docs/menu-filetree.html
b/docs/layouts/partials/docs/menu-filetree.html
index b583347dd..187b54a6d 100644
--- a/docs/layouts/partials/docs/menu-filetree.html
+++ b/docs/layouts/partials/docs/menu-filetree.html
@@ -48,6 +48,13 @@ under the License.
{{ define "book-page-link" }}{{/* (dict "Page" .Page "CurrentPage"
.CurrentPage) */}}
{{ $current := eq .CurrentPage .Page }}
{{ $ancestor := .Page.IsAncestor .CurrentPage }}
+ {{ $title := partial "docs/title" .Page }}
+ {{ if .Page.Params.icon }}
+ {{ $title = printf "%s %s" .Page.Params.icon $title }}
+ {{ end }}
+ {{ if .Page.Params.bold }}
+ {{ $title = printf `<div
style="font-weight:450;margin-bottom:0.5em">%s</div>` $title }}
+ {{ end }}
{{ if .Page.Params.sectionBreak }}
<br/>
@@ -55,14 +62,14 @@ under the License.
{{ if .Page.Params.bookCollapseSection }}
<input type="checkbox" id="section-{{ md5 .Page }}" class="toggle" {{ if
or $current $ancestor }}checked{{ end }} />
<label for="section-{{ md5 .Page }}" class="flex justify-between">
- {{- partial "docs/title" .Page -}}
+ {{- $title | safeHTML -}}
<span>▾</span>
</label>
{{ else if .Page.Content }}
<a href="{{ .Page.Permalink }}" class="{{ if $current }} active{{ end }}">
- {{- partial "docs/title" .Page -}}
+ {{- $title | safeHTML -}}
</a>
{{ else }}
- <span>{{- partial "docs/title" .Page -}}</span>
+ <span>{{- $title | safeHTML -}}</span>
{{ end }}
{{ end }}
diff --git a/docs/layouts/partials/docs/title.html
b/docs/layouts/partials/docs/title.html
index f9c96daf7..f62d444ca 100644
--- a/docs/layouts/partials/docs/title.html
+++ b/docs/layouts/partials/docs/title.html
@@ -30,13 +30,5 @@ under the License.
{{ $title = .File.BaseFileName | humanize | title }}
{{ end }}
-{{ if .Params.icon }}
- {{ $title = printf "%s %s" .Params.icon $title }}
-{{ end }}
-
-{{ if .Params.bold }}
- {{ $title = printf `<div
style="font-weight:450;margin-bottom:0.5em">%s</div>` $title }}
-{{ end }}
-
-{{ return ($title | safeHTML) }}
+{{ return $title }}
diff --git a/examples/src/main/resources/input_data.txt
b/examples/src/main/resources/input_data.txt
index 0e4fae326..9a2e269bc 100644
--- a/examples/src/main/resources/input_data.txt
+++ b/examples/src/main/resources/input_data.txt
@@ -2485,7 +2485,8 @@
{"id": "B009MA34NY", "review": "love these shoes. use them to workout inside
and to work"}
{"id": "B0014F7B98", "review": "Lightweight, comfortable, supportive. Very
happy I bought these shoes for working out."}
{"id": "B000YFSR5G", "review": "Did not fit well. Was not comfortable.
Switched to \"Fruit of the loom\". Fruit of the loom are somewhat oversized but
fit well and are very comfortable."}
-{"id": "B005AGO4LU", "review": "I wear these everyday to work, the gym,
etc."}{"id": "B0058YEJ5K", "review": "Love the fit"}
+{"id": "B005AGO4LU", "review": "I wear these everyday to work, the gym, etc."}
+{"id": "B0058YEJ5K", "review": "Love the fit"}
{"id": "B0058YEJ5K", "review": "Super comfy!"}
{"id": "B0014F7B98", "review": "I have gotten so many compliments on the blue
color and look of this shoe. It is super comfortable. The grip is wonderful
for my indoor gym cross training classes. I always order a half size larger in
Nike brand shoes and thy fit perfectly as I did with this shoe."}
{"id": "B000YFSR5G", "review": "I only give four stars because the pants have
no pockets."}
diff --git
a/python/flink_agents/e2e_tests/e2e_tests_integration/flink_integration_agent.py
b/python/flink_agents/e2e_tests/e2e_tests_integration/flink_integration_agent.py
index a90145fdb..c95307850 100644
---
a/python/flink_agents/e2e_tests/e2e_tests_integration/flink_integration_agent.py
+++
b/python/flink_agents/e2e_tests/e2e_tests_integration/flink_integration_agent.py
@@ -171,6 +171,15 @@ class TableAgent(Agent):
ctx.send_event(OutputEvent(output=content))
+class ScalarOutputAgent(Agent):
+ """Emit a scalar unchanged through the real action/output serialization
path."""
+
+ @action(EventType.InputEvent)
+ @staticmethod
+ def emit(event: Event, ctx: RunnerContext) -> None:
+
ctx.send_event(OutputEvent(output=InputEvent.from_event(event).input["value"]))
+
+
class DataStreamToTableAgent(Agent):
"""Agent used for explaining integrating agents from table to datastream.
diff --git
a/python/flink_agents/e2e_tests/e2e_tests_integration/flink_intergration_test.py
b/python/flink_agents/e2e_tests/e2e_tests_integration/flink_intergration_test.py
index 2bc1887cc..a6593e9eb 100644
---
a/python/flink_agents/e2e_tests/e2e_tests_integration/flink_intergration_test.py
+++
b/python/flink_agents/e2e_tests/e2e_tests_integration/flink_intergration_test.py
@@ -18,8 +18,10 @@
import json
import os
import sysconfig
+from collections import Counter
from pathlib import Path
+import pytest
from pyflink.common import Configuration, Encoder, WatermarkStrategy
from pyflink.common.typeinfo import BasicTypeInfo, ExternalTypeInfo,
RowTypeInfo, Types
from pyflink.datastream import (
@@ -32,6 +34,7 @@ from pyflink.datastream.connectors.file_system import (
StreamingFileSink,
)
from pyflink.table import DataTypes, Schema, StreamTableEnvironment,
TableDescriptor
+from pyflink.table.types import DataType
from flink_agents.api.execution_environment import AgentsExecutionEnvironment
from flink_agents.e2e_tests.e2e_tests_integration.flink_integration_agent
import (
@@ -39,6 +42,7 @@ from
flink_agents.e2e_tests.e2e_tests_integration.flink_integration_agent import
DataStreamToTableAgent,
ItemData,
MyKeySelector,
+ ScalarOutputAgent,
TableAgent,
)
from flink_agents.e2e_tests.test_utils import check_result
@@ -48,6 +52,39 @@ current_dir = Path(__file__).parent
os.environ["PYTHONPATH"] = sysconfig.get_paths()["purelib"]
[email protected]("table_input", [False, True], ids=["datastream",
"table"])
[email protected](
+ ("values", "data_type"),
+ [
+ ([42, -7, 0, None], DataTypes.INT()),
+ ([3.5, 0.0, None], DataTypes.DOUBLE()),
+ ([True, False, None], DataTypes.BOOLEAN()),
+ (["hello", "", None], DataTypes.STRING()),
+ ],
+ ids=["int", "float", "bool", "string"],
+)
+def test_scalar_output_to_single_column_table(
+ values: list, data_type: DataType, table_input: bool
+) -> None:
+ config = Configuration()
+ config.set_string("restart-strategy.type", "disable")
+ env = StreamExecutionEnvironment.get_execution_environment(config)
+ env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
+ env.set_parallelism(1)
+ t_env = StreamTableEnvironment.create(env)
+ agents_env = AgentsExecutionEnvironment.get_execution_environment(env,
t_env)
+ if table_input:
+ source = t_env.from_elements([(value,) for value in values], ["value"])
+ builder = agents_env.from_table(source, key_selector=lambda _: "key")
+ else:
+ source = env.from_collection([{"value": value} for value in values])
+ builder = agents_env.from_datastream(source, key_selector=lambda _:
"key")
+ schema = Schema.new_builder().column("result", data_type).build()
+ table = builder.apply(ScalarOutputAgent()).to_table(schema=schema)
+ with table.execute().collect() as results:
+ assert Counter(row[0] for row in results) == Counter(values)
+
+
def test_from_datastream_to_datastream(tmp_path: Path) -> None:
config = Configuration()
config.set_string("state.backend.type", "rocksdb")
diff --git
a/python/flink_agents/examples/quickstart/workflow_single_agent_example.py
b/python/flink_agents/examples/quickstart/workflow_single_agent_example.py
index d05b328ef..38ed777c4 100644
--- a/python/flink_agents/examples/quickstart/workflow_single_agent_example.py
+++ b/python/flink_agents/examples/quickstart/workflow_single_agent_example.py
@@ -46,6 +46,7 @@ def main() -> None:
"""
# Set up the Flink streaming environment and the Agents execution
environment.
env = StreamExecutionEnvironment.get_execution_environment()
+ env.set_parallelism(1)
agents_env = AgentsExecutionEnvironment.get_execution_environment(env)
# limit async request to avoid overwhelming ollama server
diff --git a/python/flink_agents/runtime/output_type_utils.py
b/python/flink_agents/runtime/output_type_utils.py
index 0e08ddbe0..58bafb014 100644
--- a/python/flink_agents/runtime/output_type_utils.py
+++ b/python/flink_agents/runtime/output_type_utils.py
@@ -263,20 +263,31 @@ def row_shape(row_type_info: RowTypeInfo) -> tuple:
def to_row(value: Any, shape: tuple) -> Any:
"""Adapt an emitted value to a physical ``Row`` using a :func:`row_shape`.
+ A scalar output supplies the only column of a single-column row. This
+ wrapping applies only to the outer output row, so a null nested row remains
+ null rather than becoming a row containing null.
+
The raw agent output crosses a JSON boundary, so a structured value arrives
as a dict (a Pydantic model / dataclass degraded to its fields) or, for a
``NamedTuple``, as a positional array; a value the agent emitted as a
``Row``
is reconstructed as one. Only fields the shape marks as nested rows are
- converted recursively; every other value -- scalar, map, or list -- is left
- untouched for its own coder, so no column kind is silently reshaped. This
is
- the payload of the Table/Row conversion operator.
+ converted recursively; their scalar, map, and list fields are left
untouched
+ for their own coders. This is the payload of the Table/Row conversion
operator.
"""
+ names, _ = shape
+ if len(names) == 1 and not isinstance(value, Row | dict | list | tuple):
+ return Row(value)
+ return _to_row(value, shape)
+
+
+def _to_row(value: Any, shape: tuple) -> Any:
+ """Convert structured row values without wrapping scalar or null fields."""
if isinstance(value, Row):
return value
names, children = shape
if isinstance(value, dict):
fields = [
- to_row(value.get(name), child) if child is not None else
value.get(name)
+ _to_row(value.get(name), child) if child is not None else
value.get(name)
for name, child in zip(names, children, strict=False)
]
return Row(*fields)
@@ -286,7 +297,7 @@ def to_row(value: Any, shape: tuple) -> Any:
# order. A length mismatch means the sequence is one column's own value
# (for example an ARRAY), which is left untouched for its coder.
fields = [
- to_row(item, child) if child is not None else item
+ _to_row(item, child) if child is not None else item
for item, child in zip(value, children, strict=False)
]
return Row(*fields)
diff --git a/python/flink_agents/runtime/tests/test_output_type_utils.py
b/python/flink_agents/runtime/tests/test_output_type_utils.py
index 67d5d137e..60c493530 100644
--- a/python/flink_agents/runtime/tests/test_output_type_utils.py
+++ b/python/flink_agents/runtime/tests/test_output_type_utils.py
@@ -218,12 +218,39 @@ def test_to_row_passes_through_row():
assert to_row(existing, shape) is existing
-def test_to_row_passes_through_non_dict():
- # A non-mapping value is left untouched for its own coder.
[email protected]("value", [42, -7, 3.5, True, False, "hello", "",
None])
+def test_to_row_wraps_scalar_in_single_column(value):
+ # Exercise the same JSON boundary as an agent's OutputEvent.
+ event = OutputEvent(output=value)
+ degraded = Event.from_json(event.model_dump_json()).attributes["output"]
+ shape = row_shape(RowTypeInfo([Types.STRING()], ["result"]))
+ assert to_row(degraded, shape) == Row(value)
+
+
+def test_to_row_does_not_wrap_scalar_for_multiple_columns():
shape = row_shape(infer_row_type_info(ModelOutput))
assert to_row(42, shape) == 42
[email protected]("nested", [None, {"value": None}, {"value": 42}])
+def test_to_row_preserves_nested_row_and_collection_fields(nested):
+ shape = row_shape(
+ RowTypeInfo(
+ [
+ RowTypeInfo([Types.INT()], ["value"]),
+ Types.MAP(Types.STRING(), Types.INT()),
+ Types.LIST(Types.INT()),
+ ],
+ ["nested", "mapping", "items"],
+ )
+ )
+ mapping, items = {"a": 1}, [1, 2]
+ converted = to_row({"nested": nested, "mapping": mapping, "items": items},
shape)
+ assert converted[0] == (None if nested is None else Row(nested["value"]))
+ assert converted[1] is mapping
+ assert converted[2] is items
+
+
def test_to_row_named_tuple_through_output_event_boundary():
"""A ``NamedTuple`` degrades to a positional array across the JSON
boundary.
diff --git
a/python/flink_agents/runtime/tests/test_remote_execution_environment.py
b/python/flink_agents/runtime/tests/test_remote_execution_environment.py
index 5256fffa7..2f9b6f88a 100644
--- a/python/flink_agents/runtime/tests/test_remote_execution_environment.py
+++ b/python/flink_agents/runtime/tests/test_remote_execution_environment.py
@@ -562,6 +562,32 @@ def test_to_table_schema_only_builds_row_stream():
assert t_env.from_data_stream.call_args.args[1] is schema
[email protected]("use_schema", [False, True])
+def test_scalar_table_view_preserves_scalar_datastream_views(use_schema):
+ schema = (
+ Schema.new_builder()
+ .column("result", DataTypes.INT())
+ .column_by_expression("doubled", "result * 2")
+ .build()
+ )
+ with _applied_builder() as (builder, raw_mock, _, _):
+ builder.to_datastream()
+ builder.to_datastream(Types.INT())
+ if use_schema:
+ builder.to_table(schema=schema)
+ else:
+ builder.to_table(output_type=Types.ROW_NAMED(["result"],
[Types.INT()]))
+ payload = cloudpickle.dumps(42)
+ raw_view, scalar_view, row_view = raw_mock.map.call_args_list
+ assert raw_view.args[0](payload) == 42
+ assert scalar_view.args[0](payload) == 42
+ assert row_view.args[0](payload) == Row(42)
+ assert isinstance(row_view.kwargs["output_type"], ExternalTypeInfo)
+ # The worker closure remains picklable; only physical columns consume
values.
+ restored = cloudpickle.loads(cloudpickle.dumps(row_view.args[0]))
+ assert restored(payload) == Row(42)
+
+
def test_to_table_output_type_derives_schema():
with _applied_builder() as (builder, _, t_env, _):
builder.to_table(output_type=ModelOutput)
