Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
11 changes: 5 additions & 6 deletions docs/content/docs/development/calling-llms/chat-models/_index.md
Original file line number Diff line number Diff line change
Expand Up @@ -51,12 +51,11 @@ The `@chat_model_setup` decorator or `@ChatModelSetup` annotation marks a method

Flink Agents supports cross-language chat model integration, allowing you to use chat models implemented in one language (Java or Python) from agents written in the other language. This is particularly useful when a chat model provider is only available in one language (e.g., the DashScope integration for Alibaba Cloud Model Studio is currently Python-only).

{{< hint warning >}}
**Limitations:**

- Cross-language resources are currently supported only when [running in Flink]({{< ref "docs/operations/deployment#run-in-flink" >}}), not in local development mode
- Complex object serialization between languages may have limitations
{{< /hint >}}
Choose a Java or Python provider implementation and install its dependencies in the
job's environment, following [the setup instructions]({{< ref "docs/development/create-agent/choose-apis-and-languages#integrate-the-agent-into-a-flink-job" >}}).
When provider results are carried in Event payloads, those payloads must use the supported
[Java and Python data types]({{< ref "docs/development/create-agent/choose-apis-and-languages#data-passed-between-java-and-python" >}}).
For async calls, also check the [cross-language execution requirements]({{< ref "docs/development/create-agent/actions#cross-language-async-requirements" >}}).

### How To Use

Expand Down
4 changes: 4 additions & 0 deletions docs/content/docs/development/calling-llms/mcp.md
Original file line number Diff line number Diff line change
Expand Up @@ -223,6 +223,10 @@ Flink Agents offers two implementations of MCP support, based on MCP SDKs in dif
| Java | JDK 16 and below | Python SDK |


When the selected implementation uses Python, install its Python dependencies and
configure the Python environment even for a Java Flink job; see
[dependency setup]({{< ref "docs/development/create-agent/choose-apis-and-languages#integrate-the-agent-into-a-flink-job" >}}).

As shown in the table above, for Java agents running on JDK 17+, the framework automatically uses the Java SDK implementation. If you need to use the Python SDK instead (not recommended), you can set the `lang` parameter to `"python"` in the `@MCPServer` annotation:
```java
@MCPServer(lang = "python")
Expand Down
6 changes: 3 additions & 3 deletions docs/content/docs/development/calling-llms/skills.md
Original file line number Diff line number Diff line change
Expand Up @@ -310,11 +310,11 @@ This keeps each request lean — a skill the agent never activates costs only it

The complete example sends arithmetic questions such as `What is (2 + 3) * 4?` to `MathAgent`. Its model loads `math-calculator`, uses `bash` to run `bc`, and returns the numeric result through the normal chat response path. For the chat request and response Actions, see [Chat with Models]({{< ref "docs/development/calling-llms/chat" >}}).

| Language | Flink job entry point | Agent definition | Skill instructions |
|----------|-----------------------|------------------|--------------------|
| Language | Example job | Agent definition | Skill instructions |
|----------|-------------|------------------|--------------------|
| Python | [skills_agent_example.py](https://github.com/apache/flink-agents/blob/main/python/flink_agents/examples/quickstart/skills_agent_example.py) | [MathAgent](https://github.com/apache/flink-agents/blob/main/python/flink_agents/examples/quickstart/agents/math_agent.py) | [SKILL.md](https://github.com/apache/flink-agents/blob/main/python/flink_agents/examples/quickstart/resources/skills/math-calculator/SKILL.md) |
| Java | [SkillsAgentExample.java](https://github.com/apache/flink-agents/blob/main/examples/src/main/java/org/apache/flink/agents/examples/SkillsAgentExample.java) | [MathAgent](https://github.com/apache/flink-agents/blob/main/examples/src/main/java/org/apache/flink/agents/examples/agents/MathAgent.java) | [SKILL.md](https://github.com/apache/flink-agents/blob/main/examples/src/main/resources/skills/math-calculator/SKILL.md) |

The Python example loads its bundled skill through `Skills.from_package`; the Java example uses `Skills.fromClasspath`. Make `bash` and `bc` available on the TaskManagers and prepare the Ollama model configured by the example (`qwen3.5:9b`).

Follow [Setup]({{< ref "docs/get-started/setup" >}}) for dependencies and [Build Your First Agent]({{< ref "docs/get-started/quickstart/create-agent/single-agent#run-the-example" >}}) for the build and submission workflow, substituting the Skills job entry point above. See [Integrate with Flink]({{< ref "docs/development/integrate_with_flink" >}}) for stream integration details.
Follow [Setup]({{< ref "docs/get-started/setup" >}}) for dependencies and [Build Your First Agent]({{< ref "docs/get-started/quickstart/create-agent/single-agent#run-the-example" >}}) for the build and submission workflow, using the corresponding Skills example job above. See [Integrate with Flink]({{< ref "docs/development/integrate_with_flink" >}}) for stream integration details.
1 change: 1 addition & 0 deletions docs/content/docs/development/create-agent/_index.md
Original file line number Diff line number Diff line change
Expand Up @@ -26,4 +26,5 @@ under the License.
- [Overview]({{< ref "docs/development/create-agent/overview" >}})
- [Actions]({{< ref "docs/development/create-agent/actions" >}})
- [Events]({{< ref "docs/development/create-agent/events" >}})
- [Choose APIs and Languages]({{< ref "docs/development/create-agent/choose-apis-and-languages" >}})
- [YAML API]({{< ref "docs/development/create-agent/yaml" >}})
164 changes: 108 additions & 56 deletions docs/content/docs/development/create-agent/actions.md
Original file line number Diff line number Diff line change
Expand Up @@ -243,6 +243,72 @@ evaluated as doubles and may lose precision.
start. This happens before event-time evaluation and is not controlled by the condition evaluation
failure strategy. Dynamic access inside a selected top-level attribute remains supported.

### Cross-language Actions

To declare a cross-language Action, apply `action(...)` to a `Function` descriptor
in Python, or annotate a `static final` `Function` field in Java. The descriptor
identifies the function to execute.

{{< tabs "Cross-language Actions" >}}

{{< tab "Python" >}}
```python
from flink_agents.api.function import JavaFunction

class MyAgent(Agent):
# Action signatures are fixed (Event, RunnerContext), so for_action
# fills the Java parameter types for you — only the class and method.
handle_input = action(InputEvent.EVENT_TYPE)(
JavaFunction.for_action("com.example.MyHandlers", "handleInput"))
```
{{< /tab >}}

{{< tab "Java" >}}
```java
import org.apache.flink.agents.api.function.PythonFunction;

public class MyAgent extends Agent {
@Action(EventType.InputEvent)
private static final PythonFunction handleInput =
PythonFunction.of("my_pkg.handlers", "handle_input");
}
```
{{< /tab >}}

{{< /tabs >}}

**Override the Action name (optional)**

By default, the examples above use the member name as the Action name:
`handle_input` in Python and `handleInput` in Java. Set `name` to give the Action
a different name. These declarations name the Action `process_input`:

{{< tabs "Override Cross-language Action Name" >}}

{{< tab "Python" >}}
```python
class MyAgent(Agent):
handle_input = action(InputEvent.EVENT_TYPE, name="process_input")(
JavaFunction.for_action("com.example.MyHandlers", "handleInput")
)
```
{{< /tab >}}

{{< tab "Java" >}}
```java
public class MyAgent extends Agent {
@Action(value = EventType.InputEvent, name = "process_input")
private static final PythonFunction handleInput =
PythonFunction.of("my_pkg.handlers", "handle_input");
}
```
{{< /tab >}}

{{< /tabs >}}

For dependency setup and supported payload types, see
[Choose APIs and Languages]({{< ref "docs/development/create-agent/choose-apis-and-languages" >}}).

### Async Execution

Use `execute_async` (Python) or `executeAsync` (Java) to run a function
Expand Down Expand Up @@ -296,6 +362,48 @@ Obtain resources before creating calls, and update memory or send events after
awaiting results. Callbacks must not access `RunnerContext`, and any clients they
share must support concurrent use.

#### JDK and scheduling requirements

For concurrency between Actions:

- **Java on JDK 21+:** enable continuation-based execution by adding
`--add-exports=java.base/jdk.internal.vm=ALL-UNNAMED` to the JVM options, including
`env.java.opts.all` when running on a cluster.
- **Java on JDK 11–20:** the experimental parallel execution engine is enabled by
default for plans whose Actions are all Java. If the plan declares Python Resources,
it also requires a Flink version supporting cross-language async calls (below).
Setting `parallel-execution.enabled` to `false`, or including Python Actions,
disables this engine; a Java single-call `executeAsync(...).await()` then runs
synchronously. Use `gather` to run a batch concurrently.
- **Python Actions:** use `async def` and the framework's `AsyncFuture`/`gather` APIs.
Use `gather` to await multiple calls concurrently; awaiting each call in turn runs
them sequentially.

See [Core Options]({{< ref "docs/operations/configuration#core-options" >}})
for worker-pool and concurrency limits. These scheduling rules also apply to durable
async calls; persistence adds the [durable execution requirements](#durable-execution).

#### Cross-language async requirements

Async calls across Java/Python require Flink **1.20.5+**, **2.0.2+**, **2.1.3+**,
**2.2.1+**, or **2.3+**, within the respective release line. Use these versions for
custom cross-language async calls and for asynchronous built-in chat and
context-retrieval Actions that use an opposite-language Resource.

{{< hint warning >}}
Async actions in either language may call cross-language resources. Within one async execution,
keep the cross-language call chain to one reverse callback. Only the following call shapes are
supported:

- Python action → Java resource → Python callback
- Java action → Python resource → Java callback

For example, a Python action may use a Java `ChatModelSetup` backed by a Python
`ChatModelConnection`. Multiple or deeper cross-language callbacks, such as
Python→Java→Python→Java or Java→Python→Java→Python, are not recommended and may deadlock
when an outer call occupies an async or interpreter worker while waiting for another worker.
{{< /hint >}}

### Durable Execution

Use durable execution when an operation needs result persistence and recovery replay. When a matching persisted result is available during recovery, the framework reuses it without invoking the function again. When recovery re-enters an action that has not been recorded as completed, code outside `durable_execute` / `durable_execute_async` will still be re-executed.
Expand Down Expand Up @@ -497,61 +605,5 @@ public static void processInput(Event event, RunnerContext ctx) throws Exception
}
```

{{< hint info >}}
To use async execution on JDK 21+, user should append jvm option `--add-exports=java.base/jdk.internal.vm=ALL-UNNAMED` to [env.java.opts.all](https://nightlies.apache.org/flink/flink-docs-stable/docs/deployment/config/#env-java-opts-all) before start the flink cluster.
{{< /hint >}}
{{< /tab >}}
{{< /tabs >}}

{{< hint warning >}}
Async actions in either language may call cross-language resources. Within one async execution,
keep the cross-language call chain to one reverse callback. Only the following call shapes are
supported:

- Python action → Java resource → Python callback
- Java action → Python resource → Java callback

For example, a Python action may use a Java `ChatModelSetup` backed by a Python
`ChatModelConnection`. Multiple or deeper cross-language callbacks, such as
Python→Java→Python→Java or Java→Python→Java→Python, are not recommended and may deadlock
when an outer call occupies an async or interpreter worker while waiting for another worker.
{{< /hint >}}

### Cross-language Actions

An action declared in one language can dispatch its body to the other language. The declared member *is* the executable target, so no placeholder body is needed: apply `action(...)` to a `Function` descriptor in Python, or annotate a `static final` `Function` field in Java. The member name becomes the action name, and can be overridden with `name=` (Python) or `name` (Java).

{{< tabs "Cross-language Actions" >}}

{{< tab "Python" >}}
```python
from flink_agents.api.function import JavaFunction

class MyAgent(Agent):
# Action signatures are fixed (Event, RunnerContext), so for_action
# fills the Java parameter types for you — only the class and method.
handle_input = action(InputEvent.EVENT_TYPE)(
JavaFunction.for_action("com.example.MyHandlers", "handleInput"))
```
{{< /tab >}}

{{< tab "Java" >}}
```java
import org.apache.flink.agents.api.function.PythonFunction;

public class MyAgent extends Agent {
@Action(EventType.InputEvent)
private static final PythonFunction handleInput =
PythonFunction.of("my_pkg.handlers", "handle_input");
}
```
{{< /tab >}}

{{< /tabs >}}

{{< hint warning >}}
**Limitations:**

- Cross-language actions are currently supported only when [running in Flink]({{< ref "docs/operations/deployment#run-in-flink" >}}), not in local development mode
- Complex object serialization between languages may have limitations
{{< /hint >}}
Loading
Loading