Skip to content

[python] Remove the Python local execution path #882

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
weiqingy wants to merge 5 commits into
base: main
Choose a base branch
from
Open

[python] Remove the Python local execution path #882

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
22e5d44
[python] Remove local-path tests and migrate remaining tests off it
weiqingy Jul 7, 2026
381cce2
[python] Delete the local execution path and require a Flink environment
weiqingy Jul 7, 2026
c614e84
[docs] Remove the stale local-mode configuration example
weiqingy Jul 8, 2026
fd4b425
[python] Move LocalMemoryObject into the runtime test package
weiqingy Jul 8, 2026
12d1665
[python] Re-trigger CI
weiqingy Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions docs/content/docs/development/memory/long_term_memory.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,12 +56,15 @@ When all three options are configured, the framework automatically creates a Mem
{{< tab "Python" >}}

```python
from pyflink.datastream import StreamExecutionEnvironment

from flink_agents.api.execution_environment import AgentsExecutionEnvironment
from flink_agents.api.core_options import AgentConfigOptions
from flink_agents.api.memory.long_term_memory import LongTermMemoryOptions

env = AgentsExecutionEnvironment.get_execution_environment()
agents_config = env.get_config()
env = StreamExecutionEnvironment.get_execution_environment()
agents_env = AgentsExecutionEnvironment.get_execution_environment(env)
agents_config = agents_env.get_config()

# Set job identifier (maps to Mem0 user_id)
agents_config.set(AgentConfigOptions.JOB_IDENTIFIER, "my_job")
Expand Down Expand Up @@ -455,6 +458,8 @@ The snippets below show how to read from and write to Long-Term Memory inside an
{{< tab "Python" >}}

```python
from pyflink.datastream import StreamExecutionEnvironment

from flink_agents.api.decorators import action
from flink_agents.api.execution_environment import AgentsExecutionEnvironment
from flink_agents.api.core_options import AgentConfigOptions
Expand Down Expand Up @@ -488,8 +493,9 @@ class PersonalizedAssistant:
ctx.send_event(OutputEvent(output=response))

# Setup
env = AgentsExecutionEnvironment.get_execution_environment()
agents_config = env.get_config()
env = StreamExecutionEnvironment.get_execution_environment()
agents_env = AgentsExecutionEnvironment.get_execution_environment(env)
agents_config = agents_env.get_config()
agents_config.set(AgentConfigOptions.JOB_IDENTIFIER, "personalized_assistant")
agents_config.set(LongTermMemoryOptions.Mem0.CHAT_MODEL_SETUP, "my_chat_model")
agents_config.set(LongTermMemoryOptions.Mem0.EMBEDDING_MODEL_SETUP, "my_embedding_model")
Expand Down
12 changes: 3 additions & 9 deletions docs/content/docs/operations/configuration.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,8 @@ Users can explicitly modify the configuration when defining the `AgentsExecution

```python
# Get Flink Agents execution environment
agents_env = AgentsExecutionEnvironment.get_execution_environment()
env = StreamExecutionEnvironment.get_execution_environment()
agents_env = AgentsExecutionEnvironment.get_execution_environment(env)

# Get configuration object from the environment
config = agents_env.get_configuration()
Expand Down Expand Up @@ -104,7 +105,7 @@ By default, the configuration is automatically loaded from `$FLINK_HOME/conf/con

**Special Condition**

In the following two cases, Flink Agents may not locate the corresponding configuration file, necessitating manual configuration. If the files are not set, no configuration files will be loaded, potentially resulting in unexpected behavior or failures.
In the following case, Flink Agents may not locate the corresponding configuration file, necessitating manual configuration. If the file is not set, no configuration file will be loaded, potentially resulting in unexpected behavior or failures.

- **For MiniCluster**:
Manual setup is **required** — always export the environment variable before running the job:
Expand All @@ -115,13 +116,6 @@ In the following two cases, Flink Agents may not locate the corresponding config

This ensures that Flink can locate and load the configuration file correctly.

- **Local mode**: When [run without flink]({{< ref "docs/operations/deployment">}}), use the `AgentsExecutionEnvironment.get_configuration()` API to load the YAML file directly:

```python
config = agents_env.get_configuration("path/to/your/config.yaml")
```


## Built-in configuration options

### Core Options
Expand Down
85 changes: 4 additions & 81 deletions docs/content/docs/operations/deployment.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,88 +24,11 @@ under the License.

## Overview

We provide two options to run the job:

- **Run without Flink**
- **Language Support**: Only Python
- **Input and Output**: Python List
- **Suitable Use Case**: Local Testing and Debugging

- **Run in Flink**
- **Language Support**: Python & Java
- **Input and Output**: DataStream or Table
- **Suitable Use Case**: Production

These deployment modes differ in supported languages and data formats, allowing you to choose the one that best fits your use case.

## Run without Flink

After completing the [installation of flink-agents]({{< ref "docs/get-started/installation" >}}) and building your [ReAct Agent]({{< ref "docs/development/react_agent" >}}) or [Workflow Agent]({{< ref "docs/development/workflow_agent" >}}), you can test and execute your agent locally using a simple Python script. This allows you to validate logic without requiring a Flink cluster.

### Example for Local Run with Test Data

```python
from flink_agents.api.execution_environment import AgentsExecutionEnvironment
from my_module.agents import MyAgent # Replace with your actual agent path

if __name__ == "__main__":
# 1. Initialize environment
env = AgentsExecutionEnvironment.get_execution_environment()

# 2. Prepare test data
input_data = [
{"key": "0001", "value": "Calculate the sum of 1 and 2."},
{"key": "0002", "value": "Tell me a joke about cats."}
]

# 3. Create agent instance
agent = MyAgent()

# 4. Build pipeline
output_data = env.from_list(input_data) \
.apply(agent) \
.to_list()

# 5. Execute and show results
env.execute()

print("\nExecution Results:")
for record in output_data:
for key, value in record.items():
print(f"{key}: {value}")
Flink Agents jobs run on Flink:

```

#### Input Data Format

The input data should be a list of dictionaries `List[Dict[str, Any]]` with the following structure:

```python
[
{
# Optional field: Input key.
# The key is randomly generated if not provided.
"key": "key_1",

# Required field: Input content
# This becomes the `input` field in InputEvent
"value": "Calculate the sum of 1 and 2.",
},
...
]
```

#### Output Data Format

The output data is a list of dictionaries `List[Dict[str, Any]]` where each dictionary contains a single key-value pair representing the processed result. The structure is generated from `OutputEvent` objects:

```python
[
{key_1: output_1}, # From first OutputEvent
{key_2: output_2}, # From second OutputEvent
...
]
```
- **Language Support**: Python & Java
- **Input and Output**: DataStream or Table
- **Suitable Use Case**: Production

## Run in Flink

Expand Down
5 changes: 3 additions & 2 deletions python/flink_agents/api/agents/react_agent.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,11 +62,12 @@ class OutputData(BaseModel):
result: int


env = AgentsExecutionEnvironment.get_execution_environment()
env = StreamExecutionEnvironment.get_execution_environment()
agents_env = AgentsExecutionEnvironment.get_execution_environment(env)

# register resource to execution environment
(
env.add_resource(
agents_env.add_resource(
"ollama",
ResourceDescriptor(clazz=OllamaChatModelConnection, model=model),
)
Expand Down
113 changes: 43 additions & 70 deletions python/flink_agents/api/execution_environment.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,18 +51,6 @@ def apply(self, agent: "Agent | str") -> "AgentBuilder":
on the environment (e.g. by ``load_yaml``).
"""

@abstractmethod
def to_list(self) -> List[Dict[str, Any]]:
"""Get output list of agent execution.

The element in the list is a dict like {'key': output}.

Returns:
-------
list
Outputs of agent execution.
"""

@abstractmethod
def to_datastream(self) -> DataStream:
"""Get output datastream of agent execution.
Expand Down Expand Up @@ -119,56 +107,57 @@ def get_execution_environment(
) -> "AgentsExecutionEnvironment":
"""Get agents execution environment.

Currently, user can run flink agents in ide using LocalExecutionEnvironment or
RemoteExecutionEnvironment. To distinguish which environment to use, when run
flink agents with pyflink datastream/table, user should pass flink
StreamExecutionEnvironment when get AgentsExecutionEnvironment.
A Flink ``StreamExecutionEnvironment`` is required. When running flink agents
with pyflink datastream/table, pass the ``StreamExecutionEnvironment`` so the
agents run on the Flink runtime.

Parameters
----------
env : StreamExecutionEnvironment
The Flink stream execution environment the agents run on. Must not be None.

Returns:
-------
AgentsExecutionEnvironment
Environment for agent execution.
"""
if env is None:
return importlib.import_module(
"flink_agents.runtime.local_execution_environment"
).create_instance(env=env, t_env=t_env, **kwargs)
err_msg = "A StreamExecutionEnvironment is required."
raise ValueError(err_msg)

major_version = flink_version_manager.major_version
if not major_version:
err_msg = "Apache Flink is not installed."
raise ModuleNotFoundError(err_msg)

lib_base = files("flink_agents.lib")

# Load the common JAR (shared dependencies)
common_lib = lib_base / "common"
if common_lib.is_dir():
for jar_file in common_lib.iterdir():
if jar_file.is_file() and str(jar_file).endswith(".jar"):
env.add_jars(f"file://{jar_file}")
else:
err_msg = "Flink Agents common JAR not found."
raise FileNotFoundError(err_msg)

# Load the version-specific thin JAR
version_dir = f"flink-{major_version}"
version_lib = lib_base / version_dir

# Check if version-specific directory exists
if version_lib.is_dir():
for jar_file in version_lib.iterdir():
if jar_file.is_file() and str(jar_file).endswith(".jar"):
env.add_jars(f"file://{jar_file}")
else:
major_version = flink_version_manager.major_version
if major_version:
lib_base = files("flink_agents.lib")

# Load the common JAR (shared dependencies)
common_lib = lib_base / "common"
if common_lib.is_dir():
for jar_file in common_lib.iterdir():
if jar_file.is_file() and str(jar_file).endswith(".jar"):
env.add_jars(f"file://{jar_file}")
else:
err_msg = "Flink Agents common JAR not found."
raise FileNotFoundError(err_msg)

# Load the version-specific thin JAR
version_dir = f"flink-{major_version}"
version_lib = lib_base / version_dir

# Check if version-specific directory exists
if version_lib.is_dir():
for jar_file in version_lib.iterdir():
if jar_file.is_file() and str(jar_file).endswith(".jar"):
env.add_jars(f"file://{jar_file}")
else:
err_msg = (
f"Flink Agents dist JAR for Flink {major_version} not found."
)
raise FileNotFoundError(err_msg)

return importlib.import_module(
"flink_agents.runtime.remote_execution_environment"
).create_instance(env=env, t_env=t_env, **kwargs)
else:
err_msg = "Apache Flink is not installed."
raise ModuleNotFoundError(err_msg)
err_msg = f"Flink Agents dist JAR for Flink {major_version} not found."
raise FileNotFoundError(err_msg)

return importlib.import_module(
"flink_agents.runtime.remote_execution_environment"
).create_instance(env=env, t_env=t_env, **kwargs)

@abstractmethod
def get_config(self, path: str | None = None) -> Configuration:
Expand All @@ -180,22 +169,6 @@ def get_config(self, path: str | None = None) -> Configuration:
The configuration for flink agents.
"""

@abstractmethod
def from_list(self, input: List[Dict[str, Any]]) -> AgentBuilder:
"""Set input for agents. Used for local execution.

Parameters
----------
input : list
Receive a list as input. The element in the list should be a dict like
{'key': Any, 'value': Any} or {'value': Any} , extra field will be ignored.

Returns:
-------
AgentBuilder
A new builder to build an agent for specific input.
"""

@abstractmethod
def from_datastream(
self, input: DataStream, key_selector: KeySelector | Callable | None = None
Expand Down
Loading
Loading