You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by we...@apache.org on 2026/10/15 12:30:59 UTC
(flink-agents) branch main updated: [docs] Improve Agent core concepts, Actions, and Events guides (#1263)
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 c8886cfa [docs] Improve Agent core concepts, Actions, and Events guides (#1263)
c8886cfa is described below
commit c8886cfa8b9376cdf481855ea9997ef90da558d4
Author: Wenjin Xie <16...@users.noreply.github.com>
AuthorDate: Thu Oct 15 20:30:53 2026 +0800
[docs] Improve Agent core concepts, Actions, and Events guides (#1263)
Generated-by: Codex CLI 0.153.4 (GPT-6)
Co-authored-by: Codex <co...@openai.com>
---
.../development/calling-llms/structured-output.md | 9 +-
.../docs/development/calling-llms/tool_use.md | 9 +-
.../docs/development/create-agent/actions.md | 1075 +++++++++++++-------
.../create-agent/choose-apis-and-languages.md | 10 +-
.../docs/development/create-agent/events.md | 324 +++---
.../docs/development/create-agent/overview.md | 284 +++---
docs/content/docs/development/create-agent/yaml.md | 41 +-
.../docs/development/memory/long_term_memory.md | 17 +-
.../memory/sensory_and_short_term_memory.md | 5 +
docs/content/docs/get-started/overview.md | 7 +
.../get-started/quickstart/create-agent/_index.md | 4 +-
.../quickstart/create-agent/multiple-agents.md | 4 +
.../quickstart/create-agent/single-agent.md | 15 +-
.../docs/get-started/quickstart/parallel_llm.md | 19 +-
.../docs/get-started/quickstart/react_agent.md | 6 +-
.../docs/get-started/quickstart/yaml_agent.md | 15 +-
docs/content/docs/get-started/setup.md | 2 +-
docs/content/docs/operations/configuration.md | 2 +-
docs/content/docs/operations/deployment.md | 13 +-
19 files changed, 1125 insertions(+), 736 deletions(-)
diff --git a/docs/content/docs/development/calling-llms/structured-output.md b/docs/content/docs/development/calling-llms/structured-output.md
index bce09d9c..8c0c58b5 100644
--- a/docs/content/docs/development/calling-llms/structured-output.md
+++ b/docs/content/docs/development/calling-llms/structured-output.md
@@ -105,7 +105,9 @@ ctx.sendEvent(
## Read the Result
-A ReAct Agent emits the parsed object as its output. In a custom action, read it from the successful `ChatResponseEvent`. For a typed class, Java returns an instance of the POJO. Python returns the model instance, or its fields as a `dict` once the event has been serialized, so validate it back into the model. For a `RowTypeInfo`, the result is a `Row`.
+A ReAct Agent emits the parsed object as its output. In a custom Action, read the
+structured output from a successful `ChatResponseEvent`, as shown below.
+For a `RowTypeInfo`, the result is a `Row`.
{{< tabs "Read the Result" >}}
@@ -138,6 +140,11 @@ public static void processResponse(Event event, RunnerContext ctx) throws Except
{{< /tabs >}}
+The structured output may sometimes arrive as a `Map` or `dict` instead of an instance
+of your application type. In that case, deserialize it into your application type
+before accessing its fields. See
+[Serialization and Typed Values]({{< ref "docs/development/create-agent/events#json-serialization" >}}).
+
## Choose a Strategy
Set `structured_output_strategy` on the chat model **setup** (not the connection). The value is case-insensitive.
diff --git a/docs/content/docs/development/calling-llms/tool_use.md b/docs/content/docs/development/calling-llms/tool_use.md
index 7d18c730..dc7fef4e 100644
--- a/docs/content/docs/development/calling-llms/tool_use.md
+++ b/docs/content/docs/development/calling-llms/tool_use.md
@@ -341,10 +341,9 @@ that subtask). On JDK 21+ and in Python, chat and RAG calls also use this pool.
Java batches use a separate pool from the Action workers. The default batch parallelism is
the host CPU count.
-Built-in actions for a **single key** run one at a time. In the usual chat → tool path, a chat
-async call finishes before `tool_call_action` starts, so they do not overlap on the same key.
-Contention appears mainly **across keys** on the same subtask: one key's parallel tool batch can
-delay another key's chat or RAG async work.
+Parallel tool calls share execution capacity with other external calls. A large tool
+batch may delay other model calls or retrieval requests. Use `async.batch.parallelism`
+to limit how many tools run concurrently.
{{< hint warning >}}
**Cross-key impact:** with defaults (`num-async-threads = 2× cores`, `async.batch.parallelism =
@@ -494,6 +493,6 @@ The built-in `tool_call_action` listens to `ToolRequestEvent`. For each tool cal
Ordinary tool return values are converted to successful text responses as described in [Tool response content and metadata](#tool-response-content-and-metadata). A tool that completes normally but cannot perform the requested operation can return `ToolResponse.error("reason")`; an exception still represents an invocation failure. Both failure forms are recorded as failed tool calls in `ToolResponseEvent`.
-When the tool request comes from `chat_model_action`, the emitted `ToolResponseEvent` is automatically consumed by `chat_model_action` to continue the chat. See [Tool Calling in Chat with Models]({{< ref "docs/development/calling-llms/chat#built-in-events-and-actions" >}}) for details on how `chat_model_action` handles tool responses.
+When the tool request comes from `chat_model_action`, that Action automatically handles the emitted `ToolResponseEvent` to continue the chat. See [Tool Calling in Chat with Models]({{< ref "docs/development/calling-llms/chat#built-in-events-and-actions" >}}) for details on how `chat_model_action` handles tool responses.
Users can also send `ToolRequestEvent` directly when they want to invoke tools programmatically.
diff --git a/docs/content/docs/development/create-agent/actions.md b/docs/content/docs/development/create-agent/actions.md
index 4ceeb2a9..fb234591 100644
--- a/docs/content/docs/development/create-agent/actions.md
+++ b/docs/content/docs/development/create-agent/actions.md
@@ -22,37 +22,38 @@ specific language governing permissions and limitations
under the License.
-->
-## Action
+An Action reacts to Events that match its trigger conditions. It receives the triggering
+`Event` and a `RunnerContext`, which it uses to access Resources, work with Memory, and
+send new Events. Several Actions can react to the same Event; handling an Event does not
+prevent other matching Actions from receiving it.
-An action is a piece of code that can be executed. It declares one or more trigger conditions and is
-triggered when an event matches one of them.
+Define an Action with `@action(...)` in Python or `@Action(...)` in Java. The function or
+method sends its output through the context, so its return type is `None` in Python or
+`void` in Java. A native Java Action must be `public static`.
-Use `@action(*trigger_conditions, name=None)` to decorate a Python function or `@Action({...})` to
-annotate a Java method. The function or method accepts the triggering `Event` and a `RunnerContext`.
-It sends events through the context rather than returning a result, so declare its return type as
-`None` in Python or `void` in Java. A native Java action must be `public static`. Python actions can
-also be defined as `async def` when using async execution (see [Async Execution](#async-execution)).
+This example accepts a review as a string, removes surrounding whitespace, and sends the
+cleaned text downstream. Imports and Flink job setup are omitted.
{{< tabs "Action Function" >}}
{{< tab "Python" >}}
```python
-class ReviewAnalysisAgent(Agent):
+class ReviewTextAgent(Agent):
@action(EventType.InputEvent)
@staticmethod
- def process_input(event: Event, ctx: RunnerContext) -> None:
- # the action logic
+ def clean_review(event: Event, ctx: RunnerContext) -> None:
+ review = InputEvent.from_event(event).input
+ ctx.send_event(OutputEvent(output=review.strip()))
```
{{< /tab >}}
{{< tab "Java" >}}
```java
-public class ReviewAnalysisAgent extends Agent {
- /** Process input event and send chat request for review analysis. */
+public class ReviewTextAgent extends Agent {
@Action(EventType.InputEvent)
- public static void processInput(Event event, RunnerContext ctx) throws Exception {
- InputEvent inputEvent = InputEvent.fromEvent(event);
- // the action logic
+ public static void cleanReview(Event event, RunnerContext ctx) {
+ String review = (String) InputEvent.fromEvent(event).getInput();
+ ctx.sendEvent(new OutputEvent(review.trim()));
}
}
```
@@ -60,550 +61,834 @@ public class ReviewAnalysisAgent extends Agent {
{{< /tabs >}}
-### Registering Actions Programmatically
+The framework creates an `InputEvent` for each input record. `OutputEvent` sends a result
+downstream; other Events can connect Actions, as shown in [Sending Events](#sending-events).
+For calls to external services, see [Async Execution](#async-execution), including how to
+write Python Actions with `async def`.
+
+By default, an Action uses its Python function name or Java method name: `clean_review`
+in the Python example and `cleanReview` in the Java example. Set `name` in the decorator
+or annotation to choose a different name, for example:
+
+| Python | Java |
+| --- | --- |
+| `@action(EventType.InputEvent, name="process_review")` | `@Action(value = EventType.InputEvent, name = "process_review")` |
-As an alternative to annotations, register an action through the `Agent` API:
+Action names must be unique within an Agent.
+
+## Registering Actions Programmatically
+
+You can also create an `Agent` instance and register a function or method with its trigger
+conditions. These examples perform the same text cleanup as above without an Agent subclass
+or an Action decorator or annotation:
{{< tabs "Programmatic Action" >}}
{{< tab "Python" >}}
```python
-agent.add_action(
- "process_event",
- [EventType.InputEvent],
- process_event,
-)
+def clean_review(event: Event, ctx: RunnerContext) -> None:
+ review = InputEvent.from_event(event).input
+ ctx.send_event(OutputEvent(output=review.strip()))
+
+
+agent = Agent()
+agent.add_action("clean_review", [EventType.InputEvent], clean_review)
```
{{< /tab >}}
{{< tab "Java" >}}
-```java
-public class ProgrammaticAgent extends Agent {
- public ProgrammaticAgent() throws NoSuchMethodException {
- addAction(
- new String[] {EventType.InputEvent},
- ProgrammaticAgent.class.getMethod(
- "processEvent", Event.class, RunnerContext.class));
- }
+Define the handler as a static method:
- public static void processEvent(Event event, RunnerContext ctx) {
- // Handle the input event.
+```java
+public class ReviewActions {
+ public static void cleanReview(Event event, RunnerContext ctx) {
+ String review = (String) InputEvent.fromEvent(event).getInput();
+ ctx.sendEvent(new OutputEvent(review.trim()));
}
}
```
+
+Then register it when setting up your job (the enclosing method must handle or declare
+`NoSuchMethodException`):
+
+```java
+Agent agent = new Agent();
+agent.addAction(
+ "clean_review",
+ new String[] {EventType.InputEvent},
+ JavaFunction.fromMethod(
+ ReviewActions.class.getMethod("cleanReview", Event.class, RunnerContext.class)),
+ null);
+```
{{< /tab >}}
{{< /tabs >}}
-### Trigger Conditions
+Both examples register the Action as `clean_review` through the first argument.
+In Java, `JavaFunction` is imported from `org.apache.flink.agents.api.function`;
+the final `null` means no Action-specific configuration is supplied.
-A trigger condition is either:
+## Trigger Conditions
-- An **exact event type**, such as `EventType.InputEvent` or a custom event's `EVENT_TYPE` constant.
-- A **condition expression** that evaluates event data to `true` or `false`. Expressions are written
- in [Common Expression Language (CEL)](https://cel.dev/); Flink Agents supports the subset described
- in [Trigger Condition Reference](#trigger-condition-reference).
+Trigger conditions determine which Events an Action reacts to.
-Conditions on the same action are alternatives (OR). For example, separate conditions for
-`EventType.InputEvent` and `attributes.urgent == true` allow any input event to trigger the action,
-even when `urgent` is false. To require both, combine them in one expression:
+### Exact Event Type
-```text
-type == EventType.InputEvent && attributes.urgent == true
-```
+The simplest way to register an Action is to specify one Event type. The Action runs
+whenever an Event with that type arrives:
+
+| Event Type | Python | Java |
+| --- | --- | --- |
+| Input records | `@action(EventType.InputEvent)` | `@Action(EventType.InputEvent)` |
+| Custom `review_ready` Events | `@action("review_ready")` | `@Action("review_ready")` |
+
+Use a built-in `EventType` constant for framework Events, or the type string you chose
+for a custom Event. No attribute check is needed when you want to handle every Event
+of that type.
+
+### Single Condition
-When passed directly to `@action` or `@Action`, `EventType.InputEvent` declares an exact event-type
-condition. Within an expression, it is a value and must be compared with `type`, as shown above. A
-bare identifier or dotted path is treated as an exact event type, so write attribute checks
-explicitly, for example `ready == true` or `attributes.ready == true`.
+To select Events by their data, use a Boolean expression as the trigger condition.
+This Action forwards an Event's attributes downstream when its `urgent` attribute is `true`:
-#### Matching Behavior
+{{< tabs "Single Condition" >}}
+
+{{< tab "Python" >}}
+```python
+class ReviewRoutingAgent(Agent):
+ @action("has(attributes.urgent) && attributes.urgent == true")
+ @staticmethod
+ def forward_review(event: Event, ctx: RunnerContext) -> None:
+ ctx.send_event(OutputEvent(output=event.attributes))
+```
+{{< /tab >}}
-When an event arrives:
+{{< tab "Java" >}}
+```java
+public class ReviewRoutingAgent extends Agent {
+ @Action("has(attributes.urgent) && attributes.urgent == true")
+ public static void forwardReview(Event event, RunnerContext ctx) {
+ ctx.sendEvent(new OutputEvent(event.getAttributes()));
+ }
+}
+```
+{{< /tab >}}
-1. The runtime selects every action with an exact event-type condition that matches the event's type.
-2. For each remaining action, it evaluates expressions in declaration order and stops at the first
- `true` result.
-3. Every selected action is executed once, even if more than one of its conditions matches.
+{{< /tabs >}}
-### Sending Events
+`has(attributes.urgent)` checks that the optional attribute exists before reading it.
+Expressions use the supported subset of
+[Common Expression Language (CEL)](https://cel.dev/).
-An action can send an event to trigger another action or emit output downstream.
+### Multiple Conditions
-**Trigger another action** — send a built-in or custom event that matches another action's trigger
-conditions:
+Conditions on the same Action are alternatives (**OR**). The following Action forwards
+an Event's attributes downstream if its type is `review_ready` **or** its `urgent`
+attribute is `true`:
-{{< tabs "Trigger Another Action" >}}
+{{< tabs "Multiple Trigger Conditions" >}}
{{< tab "Python" >}}
```python
-@action(EventType.InputEvent)
-@staticmethod
-def process_input(event: Event, ctx: RunnerContext) -> None:
- ctx.send_event(ChatRequestEvent(model="my_model", messages=messages))
+class ReviewRoutingAgent(Agent):
+ @action("review_ready", "has(attributes.urgent) && attributes.urgent == true")
+ @staticmethod
+ def forward_review(event: Event, ctx: RunnerContext) -> None:
+ ctx.send_event(OutputEvent(output=event.attributes))
```
{{< /tab >}}
{{< tab "Java" >}}
```java
-@Action(EventType.InputEvent)
-public static void processInput(Event event, RunnerContext ctx) throws Exception {
- ctx.sendEvent(new ChatRequestEvent("my_model", messages));
+public class ReviewRoutingAgent extends Agent {
+ @Action({"review_ready", "has(attributes.urgent) && attributes.urgent == true"})
+ public static void forwardReview(Event event, RunnerContext ctx) {
+ ctx.sendEvent(new OutputEvent(event.getAttributes()));
+ }
}
```
{{< /tab >}}
{{< /tabs >}}
-**Emit downstream output** — send an `OutputEvent`:
+| Event type | `urgent` attribute | Does the Action run? |
+| --- | --- | --- |
+| `review_ready` | Absent or `false` | Yes: the Event type matches. |
+| `review_updated` | `true` | Yes: the expression matches. |
+| `review_ready` | `true` | Yes, once: matching both conditions does not cause a second execution. |
+| `review_updated` | Absent or `false` | No. |
+
+To require **both** the type and the attribute to match, put them in a single expression:
+
+```text
+type == 'review_ready' && has(attributes.urgent) && attributes.urgent == true
+```
+
+### Matching Behavior
-{{< tabs "Emit Output" >}}
+For each Event, the framework selects every Action with a matching trigger condition.
+Each selected Action runs once for that Event, even if several of its conditions match.
+Different Actions can all respond to the same Event.
+
+For supported names, attribute access, and evaluation failures, see the
+[Trigger Condition Reference](#trigger-condition-reference).
+
+## Sending Events
+
+Use `ctx.send_event(...)` in Python or `ctx.sendEvent(...)` in Java to send an Event.
+The Event's type and attributes determine which Actions react to it.
+
+To split the text cleanup example into two steps, have the first Action send a
+`review_ready` Event carrying the cleaned text. A second Action listens for that type
+and sends the text downstream:
+
+{{< tabs "Connect Actions with Events" >}}
{{< tab "Python" >}}
```python
-@action(EventType.ChatResponseEvent)
-@staticmethod
-def emit_output(event: Event, ctx: RunnerContext) -> None:
- ctx.send_event(OutputEvent(output=result))
+class ReviewTextAgent(Agent):
+ @action(EventType.InputEvent)
+ @staticmethod
+ def clean_review(event: Event, ctx: RunnerContext) -> None:
+ review = InputEvent.from_event(event).input
+ ctx.send_event(
+ Event(type="review_ready", attributes={"text": review.strip()})
+ )
+
+ @action("review_ready")
+ @staticmethod
+ def emit_review(event: Event, ctx: RunnerContext) -> None:
+ ctx.send_event(OutputEvent(output=event.get_attr("text")))
```
{{< /tab >}}
{{< tab "Java" >}}
```java
-@Action(EventType.ChatResponseEvent)
-public static void emitOutput(Event event, RunnerContext ctx) {
- ctx.sendEvent(new OutputEvent(result));
+public class ReviewTextAgent extends Agent {
+ @Action(EventType.InputEvent)
+ public static void cleanReview(Event event, RunnerContext ctx) {
+ String review = (String) InputEvent.fromEvent(event).getInput();
+ ctx.sendEvent(new Event("review_ready", Map.of("text", review.trim())));
+ }
+
+ @Action("review_ready")
+ public static void emitReview(Event event, RunnerContext ctx) {
+ ctx.sendEvent(new OutputEvent(event.getAttr("text")));
+ }
}
```
{{< /tab >}}
{{< /tabs >}}
+Flink Agents automatically registers [built-in Actions]({{< ref "docs/development/create-agent/events#built-in-events-and-actions" >}})
+to handle common request Events. For example, a `ChatRequestEvent`
+asks the [built-in chat Action]({{< ref "docs/development/calling-llms/chat#chat-events" >}})
+to call a registered model and send a `ChatResponseEvent`.
+
+You can also define your own Events to pass data between Actions. See
+[Creating and Handling Events]({{< ref "docs/development/create-agent/events#unified-event" >}})
+for examples, or use [Custom Event Subclasses]({{< ref "docs/development/create-agent/events#custom-event-subclasses" >}})
+to add typed access and validation.
+
{{< hint info >}}
-`OutputEvent` is emitted directly downstream and bypasses action matching. Therefore, an
-`OutputEvent` trigger never invokes an action. Other events go through action matching as usual.
+`OutputEvent` is emitted directly downstream and bypasses Action matching. Therefore, an
+`OutputEvent` trigger never invokes an Action. Other Events go through Action matching as usual.
{{< /hint >}}
-### Trigger Condition Reference
+## Cross-language Actions
-#### Custom Event-Type Names
+An Agent can register Actions implemented in another language. A Python Agent can use
+a Java method as an Action, and a Java Agent can use a Python function as an Action.
+The handler still receives an `Event` and a `RunnerContext` and sends its output through
+the context.
-A custom event type may be a bare name such as `order.created` or `order-created`. Each dot-separated
-segment must start with an ASCII letter or underscore and may then contain ASCII letters, digits,
-underscores, or hyphens. Quote a name that contains other punctuation or would otherwise be parsed as
-an expression, for example `'order:created'`, `'true'`, or `'EventType.custom'`. In Java and Python,
-the quotes are part of the condition string. For example, use `@Action("'order:created'")` in Java or
-`@action("'order:created'")` in Python. A quoted name matches the literal event-type string; it does
-not reference a built-in `EventType` constant. Quoted names must be non-empty and cannot contain
-whitespace, quotes, backslashes, or control characters.
+The following examples perform the same text cleanup as above. Each example shows the
+handler and how to register it from the other language. Flink job setup is omitted.
-#### Expression Variables
-
-Java and Python actions use the same Java runtime to evaluate condition expressions. Expressions can
-access these framework variables:
+{{< tabs "Cross-language Actions" >}}
-| Variable | Value |
-|--------------|--------------------------------------|
-| `type` | The event type string |
-| `id` | The event ID as a string |
-| `EventType` | The built-in event-type constants |
-| `attributes` | The event's top-level attribute map |
+{{< tab "Python Agent" >}}
-#### Accessing Attributes
+Define the Java handler in `com/example/ReviewActions.java`:
-- Top-level attributes are also available as bare variables, so `score > 80` and
- `attributes.score > 80` refer to the same field.
-- Nested values are not flattened. For `{input: {status: "ok"}}`, use `input.status` or
- `attributes.input.status`; bare `status` does not refer to the nested value. Other event payloads
- keep their top-level envelope, for example `response.message.blocks`.
-- Framework variables take precedence over attributes with the same names. Use `attributes["type"]`
- or `attributes["id"]` to access a colliding attribute.
-- For a top-level key containing dots, use a literal index such as `attributes["a.b.c"]`. Test its
- presence with `"a.b.c" in attributes`.
+```java
+package com.example;
-#### Missing Attributes and Value Types
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.InputEvent;
+import org.apache.flink.agents.api.OutputEvent;
+import org.apache.flink.agents.api.context.RunnerContext;
-A present attribute whose value is `null` remains `null`, while a missing attribute remains absent.
-Use `has(attributes.field)` before reading an optional attribute. Reading a missing attribute without
-a guard causes an event-time evaluation failure. With the default `WARN_AND_SKIP` strategy, the
-runtime logs a warning, treats that condition as false, and continues with later OR conditions;
-`FAIL` fails the Flink task. See the
-[condition evaluation failure strategy]({{< ref "docs/operations/configuration#core-options" >}}).
+public class ReviewActions {
+ public static void cleanReview(Event event, RunnerContext ctx) {
+ String review = (String) InputEvent.fromEvent(event).getInput();
+ ctx.sendEvent(new OutputEvent(review.trim()));
+ }
+}
+```
-Strings remain strings even when they contain JSON. To match nested data, send a structured map or
-list instead of a JSON-encoded string. Decimal values and integers outside the signed 64-bit range are
-evaluated as doubles and may lose precision.
+Then register that method in the Python Agent. `JavaFunction.for_action` takes the fully
+qualified Java class name and method name:
-#### Limitations
+```python
+from flink_agents.api.agents.agent import Agent
+from flink_agents.api.decorators import action
+from flink_agents.api.events.event_type import EventType
+from flink_agents.api.function import JavaFunction
-- `has(...)` is the only supported CEL macro. The comprehension macros `exists`, `exists_one`,
- `all`, `filter`, and `map` are not supported.
-- Dynamic root access such as `attributes[key]` and operations over the whole `attributes` map are
- rejected when the runtime compiles the condition. Operator initialization fails, so the job cannot
- 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
+class ReviewTextAgent(Agent):
+ clean_review = action(EventType.InputEvent)(
+ JavaFunction.for_action("com.example.ReviewActions", "cleanReview")
+ )
+```
+{{< /tab >}}
-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.
+{{< tab "Java Agent" >}}
-{{< tabs "Cross-language Actions" >}}
+Define the Python handler in `my_pkg/handlers.py`:
-{{< tab "Python" >}}
```python
-from flink_agents.api.function import JavaFunction
+from flink_agents.api.events.event import Event, InputEvent, OutputEvent
+from flink_agents.api.runner_context import RunnerContext
+
-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"))
+def clean_review(event: Event, ctx: RunnerContext) -> None:
+ review = InputEvent.from_event(event).input
+ ctx.send_event(OutputEvent(output=review.strip()))
```
-{{< /tab >}}
-{{< tab "Java" >}}
+Then register that function in the Java Agent. `PythonFunction.of` takes the Python
+module name and function name. Apply `@Action` to the `static final` field:
+
```java
+import org.apache.flink.agents.api.EventType;
+import org.apache.flink.agents.api.agents.Agent;
+import org.apache.flink.agents.api.annotation.Action;
import org.apache.flink.agents.api.function.PythonFunction;
-public class MyAgent extends Agent {
+public class ReviewTextAgent extends Agent {
@Action(EventType.InputEvent)
- private static final PythonFunction handleInput =
- PythonFunction.of("my_pkg.handlers", "handle_input");
+ private static final PythonFunction cleanReview =
+ PythonFunction.of("my_pkg.handlers", "clean_review");
}
```
{{< /tab >}}
{{< /tabs >}}
-**Override the Action name (optional)**
+For cross-language Actions, the default Action name comes from the field used to
+register it: `clean_review` in Python and `cleanReview` in Java. You can override it
+with `name`, as with a native Action.
+
+For dependency setup and supported payload types, see
+[Choose APIs and Languages]({{< ref "docs/development/create-agent/choose-apis-and-languages" >}}).
+
+## Async and Durable Execution
+
+When an Action calls an external system, you can control how it waits for the call
+and whether the call's outcome is saved for recovery:
+
+- **[Async Execution](#async-execution)** lets the Action yield while waiting for external
+ work, allowing other Actions to run. It also supports running multiple calls concurrently.
+- **[Durable Execution](#durable-execution)** saves a call's outcome so it can be reused
+ during recovery.
+- **[Durable Async Execution](#durable-async-execution)** combines both capabilities.
+
+See [Execution Requirements](#execution-requirements) for runtime prerequisites.
+
+### Async Execution
+
+Use async execution for calls that spend time waiting for external systems, such as
+looking up a user profile. Async execution controls how calls run and how your Action
+waits for them. To also save their results for recovery, see [Durable Async Execution](#durable-async-execution).
-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`:
+#### Single Call
-{{< tabs "Override Cross-language Action Name" >}}
+The examples below accept a user ID as a string. `profile_service.get_profile` in Python
+and `ProfileService.getProfile` in Java represent your application's synchronous client
+function: each takes a user ID and returns a profile as a string. Supply your own client
+implementation; framework imports and job setup are omitted.
+{{< tabs "Single Async Call" >}}
{{< tab "Python" >}}
```python
-class MyAgent(Agent):
- handle_input = action(InputEvent.EVENT_TYPE, name="process_input")(
- JavaFunction.for_action("com.example.MyHandlers", "handleInput")
- )
+class ProfileAgent(Agent):
+ @action(EventType.InputEvent)
+ @staticmethod
+ async def process_input(event: Event, ctx: RunnerContext) -> None:
+ user_id = InputEvent.from_event(event).input
+ future = ctx.execute_async(profile_service.get_profile, user_id)
+ profile = await future
+ ctx.send_event(OutputEvent(output=profile))
```
{{< /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");
+public class ProfileAgent extends Agent {
+ @Action(EventType.InputEvent)
+ public static void processInput(Event event, RunnerContext ctx) throws Exception {
+ String userId = (String) InputEvent.fromEvent(event).getInput();
+ AsyncFuture<String> future = ctx.executeAsync(() -> ProfileService.getProfile(userId));
+ String profile = future.await();
+ ctx.sendEvent(new OutputEvent(profile));
+ }
}
```
{{< /tab >}}
-
{{< /tabs >}}
-For dependency setup and supported payload types, see
-[Choose APIs and Languages]({{< ref "docs/development/create-agent/choose-apis-and-languages" >}}).
+Creating an `AsyncFuture` does not start the call. Awaiting it starts the work and waits
+for its result. A future that is never awaited performs no work. Awaiting the same future
+again reuses its result or exception within the current Action execution.
-### Async Execution
+In Python, the Action uses `async def`, but the function passed to `execute_async` must
+be synchronous. Use the framework's futures and `gather`; standard `asyncio` functions
+such as `asyncio.gather` and `asyncio.create_task` are not supported.
-Use `execute_async` (Python) or `executeAsync` (Java) to run a function
-asynchronously, for example when querying an external service.
+#### Concurrent Calls
-These methods return an `AsyncFuture`. Work starts when you await the future,
-not when you create it. Await one call directly, or use `ctx.gather(...)` to combine
-multiple calls. Awaiting each call separately executes them sequentially.
+To look up a profile and preferences concurrently, create both futures and await them
+with `ctx.gather(...)`. Here, `get_preferences` / `getPreferences` is another synchronous
+application function that takes a user ID and returns a string.
-{{< tabs "Ordinary Async Execution" >}}
+{{< tabs "Concurrent Async Calls" >}}
{{< tab "Python" >}}
-Inside an `async def` action, pass synchronous functions to `execute_async`:
-
```python
-profile = ctx.execute_async(query_profile, user_id)
-inventory = ctx.execute_async(query_inventory, item_id)
-outcomes = await ctx.gather(profile, inventory)
-for outcome in outcomes:
- if outcome.is_failure():
- raise outcome.error
-profile_result = outcomes[0].value
+class ProfileAgent(Agent):
+ @action(EventType.InputEvent)
+ @staticmethod
+ async def process_input(event: Event, ctx: RunnerContext) -> None:
+ user_id = InputEvent.from_event(event).input
+ profile = ctx.execute_async(profile_service.get_profile, user_id)
+ preferences = ctx.execute_async(profile_service.get_preferences, user_id)
+ outcomes = await ctx.gather(profile, preferences)
+ for outcome in outcomes:
+ if outcome.is_failure():
+ raise outcome.error
+ ctx.send_event(OutputEvent(output=f"{outcomes[0].value}\n{outcomes[1].value}"))
```
-
-For a single call, use `result = await ctx.execute_async(query_profile, user_id)`.
-Use `ctx.gather` for composition; standard asyncio functions are not supported.
{{< /tab >}}
{{< tab "Java" >}}
```java
-AsyncFuture<String> profile = ctx.executeAsync(() -> queryProfile(userId));
-AsyncFuture<String> inventory = ctx.executeAsync(() -> queryInventory(itemId));
-List<Outcome<String>> outcomes = ctx.gather(List.of(profile, inventory)).await();
-for (Outcome<String> outcome : outcomes) {
- if (outcome.isFailure()) {
- throw outcome.getError();
+public class ProfileAgent extends Agent {
+ @Action(EventType.InputEvent)
+ public static void processInput(Event event, RunnerContext ctx) throws Exception {
+ String userId = (String) InputEvent.fromEvent(event).getInput();
+ AsyncFuture<String> profile = ctx.executeAsync(() -> ProfileService.getProfile(userId));
+ AsyncFuture<String> preferences =
+ ctx.executeAsync(() -> ProfileService.getPreferences(userId));
+ List<Outcome<String>> outcomes = ctx.gather(List.of(profile, preferences)).await();
+ for (Outcome<String> outcome : outcomes) {
+ if (outcome.isFailure()) {
+ throw outcome.getError();
+ }
+ }
+ ctx.sendEvent(
+ new OutputEvent(outcomes.get(0).getValue() + "\n" + outcomes.get(1).getValue()));
}
}
-String profileResult = outcomes.get(0).getValue();
```
-
-For a single call, use `String result = ctx.executeAsync(() -> queryProfile(userId)).await()`.
{{< /tab >}}
{{< /tabs >}}
-`gather` returns an `Outcome` for each call in input order, including individual
-failures. Batch concurrency is controlled by `async.batch.parallelism`.
-Java and Python support concurrent batches on all supported JDK versions.
-
-Keep futures within the action that created them. Repeated awaits reuse the same
-future's result or exception; a future that is never awaited performs no work.
-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 >}}
+Awaiting the futures individually, one after another, runs these calls sequentially.
+Awaiting the gathered future allows them to run concurrently, subject to
+`async.batch.parallelism` and the available worker threads.
-### Durable Execution
+#### Results and Failures
-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.
+- Awaiting a single call returns its result or throws its exception.
+- `gather` returns an `Outcome` for each call in input order, regardless of completion
+ order. An individual call's failure is represented by a failed `Outcome`. Inspect the
+ outcomes before using their values; the example propagates any individual failure.
+- A batch may itself fail, for example because of a timeout or cancellation. It does not
+ always return a list of outcomes.
-**Constraints:**
-- The function must be deterministic and called in the same order on recovery.
-- Access to Memory and `send_event` is prohibited inside the function/callable.
-- Arguments and results must be serializable.
+Keep futures within the Action execution that created them. Do not pass them to another
+Action or await them inside a submitted function. `gather` accepts distinct single-call
+futures from the same context; nested gathered futures are not supported.
-{{< hint info >}}
-Durable execution requires an external action state store. See
-[Exactly-Once Action Consistency]({{< ref "docs/operations/deployment#exactly-once-action-consistency" >}})
-on how to setup and configure the external action state store.
-{{< /hint >}}
+#### Using Resources and Memory
-**Best-effort replay:**
-- Results may not be reused if call order or arguments change (non-deterministic actions), which clears subsequent cached results and re-executes.
-- If a failure happens after a function starts but before it completes and its result is persisted, the call will be re-executed. See the "With a reconciler" section below.
-- In Python async actions, if `ctx.durable_execute_async(...)` is not awaited, the result is not recorded and cannot be replayed.
-- In Python, a recorded failure is replayed as the original exception when it can be pickled and rebuilt. Otherwise it is replayed as a `RuntimeError` with the message `<class>: <message>`, so code that catches a specific exception type may take a different path during replay. A failure that cannot be pickled is recorded this way too, so the call is not executed again on recovery. Failures recorded by a version that stored only the pickled exception have no class name or message to fall [...]
+Distinguish the **Action body**, including code before and after await, from the
+**function submitted to `execute_async` / `executeAsync`**:
-**With a reconciler:**
+| Operation | Action body | Submitted function |
+| --- | --- | --- |
+| Read or write Sensory Memory / Short-term Memory | Supported | Unsupported, including through a previously obtained `MemoryObject` |
+| Look up a Resource with `ctx.get_resource` / `ctx.getResource` | Supported | Unsupported; obtain the Resource in the Action first |
+| Send Events or invoke other `RunnerContext` operations | Supported | Unsupported |
+| Get or delete a Long-term Memory set | Supported | Unsupported |
+| Call `add`, `get`, `search`, or `delete` on a Long-term Memory `MemorySet` obtained in this Action | Supported | Supported |
-Use a reconciler for durable calls when the original call may already have completed but its result or failure has not yet been persisted, so the framework cannot determine during recovery whether the call needs to be executed again. A reconciler provides custom logic that can return the result or raise the failure for the durable call instead of re-executing the original call.
+Read the data you need and obtain Resources in the Action body, pass them to the call,
+then update Sensory/Short-term Memory or send Events after awaiting the result. Clients
+and Resources shared by concurrent calls must support concurrent use.
-- A durable call may optionally provide a reconciler that is used only during recovery, when the same durable call is revisited and no execution result has been persisted for it yet.
-- If the reconciler logic returns a result, the runtime persists and replays that recovered result.
-- If the reconciler logic raises an exception, the runtime persists and replays that recovered failure.
+For Long-term Memory, obtain a `MemorySet` in each Action that needs one. The set retains
+that Action's partition key, so its operations can run in the submitted function. For
+example, inside an async Action with Long-term Memory configured:
-{{< tabs "Durable Execution" >}}
+{{< tabs "Async Long-term Memory Search" >}}
{{< tab "Python" >}}
-Python actions can call `ctx.durable_execute(...)` to run a synchronous durable code block.
```python
-@action(EventType.InputEvent)
-@staticmethod
-def process_input(event: Event, ctx: RunnerContext) -> None:
- input_event = InputEvent.from_event(event)
- def slow_external_call(data: str) -> str:
- time.sleep(2)
- return f"Processed: {data}"
-
- # Synchronous durable execution
- result = ctx.durable_execute(slow_external_call, input_event.input)
- ctx.send_event(OutputEvent(output=result))
+memory_set = ctx.long_term_memory.get_memory_set("conversations")
+items = await ctx.execute_async(memory_set.search, query="user preferences", limit=5)
+```
+{{< /tab >}}
+{{< tab "Java" >}}
+```java
+MemorySet memorySet = ctx.getLongTermMemory().getMemorySet("conversations");
+List<MemorySetItem> items =
+ ctx.executeAsync(() -> memorySet.search("user preferences", 5, null, Map.of())).await();
```
+{{< /tab >}}
+{{< /tabs >}}
+
+Do not cache a `MemorySet` across Actions, or call `get_memory_set` / `getMemorySet` or
+`delete_memory_set` / `deleteMemorySet` inside the submitted function. See
+[Long-term Memory]({{< ref "docs/development/memory/long_term_memory" >}}) for configuration
+and the scope of memory sets.
+
+### Durable Execution
-You can also pass an optional `reconciler` callable to recover an execution outcome during recovery.
+Use durable execution when a call's result should be saved and reused during recovery.
+For example, if an Action fails after a profile lookup has completed and its result has
+been persisted, recovery can reuse that profile instead of querying the service again.
+Code outside the durable call may still run again when the Action is re-executed.
+
+Durable execution requires an external action state store. Configure it as described in
+[Exactly-Once Action Consistency]({{< ref "docs/operations/deployment#exactly-once-action-consistency" >}}).
+
+#### A Durable Call
+
+The following examples make the profile lookup from [Single Call](#single-call) durable.
+The synchronous API waits for the result before continuing the Action.
+
+{{< tabs "Synchronous Durable Call" >}}
+{{< tab "Python" >}}
```python
-@action(EventType.InputEvent)
-@staticmethod
-def process_input(event: Event, ctx: RunnerContext) -> None:
- input_event = InputEvent.from_event(event)
-
- def submit_payment(order_id: str) -> str:
- return payment_client.submit(order_id)
-
- def payment_reconciler() -> str:
- status = payment_client.get_status(input_event.input)
- if status == "SUCCEEDED":
- return payment_client.lookup_completed_payment(input_event.input)
- raise payment_client.get_failure(input_event.input)
-
- result = ctx.durable_execute(
- submit_payment,
- input_event.input,
- reconciler=payment_reconciler,
- )
- ctx.send_event(OutputEvent(output=result))
+class ProfileAgent(Agent):
+ @action(EventType.InputEvent)
+ @staticmethod
+ def process_input(event: Event, ctx: RunnerContext) -> None:
+ user_id = InputEvent.from_event(event).input
+ profile = ctx.durable_execute(
+ profile_service.get_profile,
+ user_id,
+ durable_id=f"profile:{user_id}",
+ )
+ ctx.send_event(OutputEvent(output=profile))
```
{{< /tab >}}
-
{{< tab "Java" >}}
-Java actions use `DurableCallable<T>` with `ctx.durableExecute(...)`, where `getId()` must be stable and `getResultClass()` supports recovery deserialization.
```java
-@Action(EventType.InputEvent)
-public static void processInput(Event event, RunnerContext ctx) throws Exception {
- InputEvent inputEvent = InputEvent.fromEvent(event);
- DurableCallable<String> call = new DurableCallable<>() {
- @Override
- public String getId() {
- return "slow_external_call";
- }
+class ProfileLookup implements DurableCallable<String> {
+ private final String userId;
- @Override
- public Class<String> getResultClass() {
- return String.class;
- }
+ ProfileLookup(String userId) {
+ this.userId = userId;
+ }
- @Override
- public String call() throws Exception {
- Thread.sleep(2000);
- return "Processed: " + inputEvent.getInput();
- }
- };
+ @Override
+ public String getId() {
+ return "profile:" + userId;
+ }
+
+ @Override
+ public Class<String> getResultClass() {
+ return String.class;
+ }
+
+ @Override
+ public String call() throws Exception {
+ return ProfileService.getProfile(userId);
+ }
+}
- String result = ctx.durableExecute(call);
- ctx.sendEvent(new OutputEvent(result));
+public class ProfileAgent extends Agent {
+ @Action(EventType.InputEvent)
+ public static void processInput(Event event, RunnerContext ctx) throws Exception {
+ String userId = (String) InputEvent.fromEvent(event).getInput();
+ String profile = ctx.durableExecute(new ProfileLookup(userId));
+ ctx.sendEvent(new OutputEvent(profile));
+ }
}
```
+{{< /tab >}}
+{{< /tabs >}}
-Java actions can also override `reconciler()` to recover an execution outcome during recovery.
-```java
-@Action(EventType.InputEvent)
-public static void processInput(Event event, RunnerContext ctx) throws Exception {
- InputEvent inputEvent = InputEvent.fromEvent(event);
- DurableCallable<String> call = new DurableCallable<>() {
- @Override
- public String getId() {
- return "submit_payment";
- }
+Python's `durable_id` and Java's `getId()` identify the logical operation during recovery.
+Java's `getResultClass()` provides the type needed to deserialize a saved result.
- @Override
- public Class<String> getResultClass() {
- return String.class;
- }
+#### Recovery Behavior
- @Override
- public String call() {
- return paymentClient.submit(inputEvent.getInput());
- }
+When recovery re-executes an unfinished Action and reaches a durable call:
- @Override
- public Callable<String> reconciler() {
- return () -> {
- PaymentStatus status =
- paymentClient.getStatus(inputEvent.getInput());
- if (status == PaymentStatus.SUCCEEDED) {
- return paymentClient.lookupCompletedPayment(
- inputEvent.getInput());
- }
- throw paymentClient.getFailure(inputEvent.getInput());
- };
- }
- };
+| State of the matching call | Recovery behavior |
+| --- | --- |
+| A successful result has been persisted | Return the saved result without calling the function again. |
+| A failure has been persisted | Replay the failure without calling the function again. |
+| No completed outcome has been persisted | Execute the function again, or use the reconciler for a previously recorded in-flight call. |
- String result = ctx.durableExecute(call);
- ctx.sendEvent(new OutputEvent(result));
-}
+An external operation may finish before its result is persisted. If a failure occurs in
+that interval, calling the operation again can repeat its side effects. For such
+operations, use the external service's idempotency support or a reconciler that can
+recover the existing outcome.
+
+#### Reconciling an In-flight Call
+
+A reconciler asks the external system what happened to an earlier call whose outcome
+was not persisted. It is used only when recovery revisits a matching, recorded in-flight
+call. A saved outcome is replayed directly; a new call executes normally.
+
+The reconciler must return a recovered result or raise the recovered failure. That
+outcome is persisted. Raising an ordinary exception does not ask the framework to retry
+the original operation; it records a failed outcome.
+
+For example, suppose a report service supports idempotent submission by a stable request
+ID. The application provides `report_service` / `ReportService` with three synchronous
+operations: `is_submitted` / `isSubmitted` checks for an existing request,
+`wait_for_result` / `waitForResult` waits for its completion and returns the report URL
+(or throws its failure), and `submit_and_wait` / `submitAndWait` submits idempotently and
+waits for the same outcome. Inside an Action, register a reconciler as follows:
+
+{{< tabs "Reconcile a Report Request" >}}
+{{< tab "Python" >}}
+```python
+request_id = InputEvent.from_event(event).input
+
+
+def reconcile_report() -> str:
+ if report_service.is_submitted(request_id):
+ return report_service.wait_for_result(request_id)
+ return report_service.submit_and_wait(request_id)
+
+
+report_url = ctx.durable_execute(
+ report_service.submit_and_wait,
+ request_id,
+ durable_id=f"report:{request_id}",
+ reconciler=reconcile_report,
+)
+ctx.send_event(OutputEvent(output=report_url))
+```
+{{< /tab >}}
+{{< tab "Java" >}}
+```java
+String requestId = (String) InputEvent.fromEvent(event).getInput();
+DurableCallable<String> call = new DurableCallable<>() {
+ @Override
+ public String getId() {
+ return "report:" + requestId;
+ }
+
+ @Override
+ public Class<String> getResultClass() {
+ return String.class;
+ }
+
+ @Override
+ public String call() throws Exception {
+ return ReportService.submitAndWait(requestId);
+ }
+
+ @Override
+ public Callable<String> reconciler() {
+ return () -> {
+ if (ReportService.isSubmitted(requestId)) {
+ return ReportService.waitForResult(requestId);
+ }
+ return ReportService.submitAndWait(requestId);
+ };
+ }
+};
+String reportUrl = ctx.durableExecute(call);
+ctx.sendEvent(new OutputEvent(reportUrl));
```
{{< /tab >}}
{{< /tabs >}}
+The same request ID must be reused after recovery. The service's idempotency guarantee
+also covers a retry racing with an earlier submission. If the service cannot establish
+a final outcome, the reconciler must implement the appropriate waiting or retry policy.
+
+#### Durable Call Requirements
+
+- Recovery must reach the same logical calls in the same order, with the same inputs.
+ Keep explicit durable IDs stable and distinguish operations or inputs that must not
+ reuse each other's result. An explicit ID does not make a reordered workflow safe.
+- Python derives an identity from the function and arguments when `durable_id` is omitted.
+ Supply a stable ID when your application owns one, as in the examples above.
+- The Action's control flow must be reproducible; the external function does not have to
+ produce the same result every time. Replaying the persisted result is what allows a
+ model call or external query to be reused.
+- Results must support the runtime's serialization: Java uses JSON and the declared result
+ class; Python uses pickle-compatible results. Keep inputs serializable and stable.
+- Read and write Sensory/Short-term Memory and send Events in the Action body, outside
+ the durable function and reconciler. Obtain Resources and Long-term Memory sets there
+ too; operations on an already obtained `MemorySet` may be performed in a durable call.
+
### Durable Async Execution
-Durable async execution adds result persistence and recovery replay to
-[async execution](#async-execution). Ordinary async calls may execute again if an
-action is re-executed after a failure. Use `durable_execute_async` (Python) or
-`durableExecuteAsync` (Java) when you need to reuse a saved result instead.
+Durable async execution combines [async calls](#async-execution) with
+[durable recovery](#durable-execution). Use it when you want to await external work and
+reuse its persisted outcome after a failure. The context and Memory rules above still apply.
-Await the returned `DurableFuture` directly or combine calls with `ctx.gather(...)`.
-A batch can mix ordinary and durable calls; only durable calls save their results
-for recovery. The [durable execution constraints](#durable-execution) also apply.
+Reuse the profile client and Java `ProfileLookup` from [A Durable Call](#a-durable-call):
-{{< tabs "Async Execution" >}}
+{{< tabs "Durable Async Call" >}}
{{< tab "Python" >}}
-Define an `async def` action and `await ctx.durable_execute_async(...)`. The same optional `reconciler=...` argument is available for recovery.
```python
-@action(EventType.InputEvent)
-@staticmethod
-async def process_with_async(event: Event, ctx: RunnerContext) -> None:
- input_event = InputEvent.from_event(event)
- def slow_external_call(data: str) -> str:
- time.sleep(2)
- return f"Processed: {data}"
-
- result = await ctx.durable_execute_async(slow_external_call, input_event.input)
- ctx.send_event(OutputEvent(output=result))
+class ProfileAgent(Agent):
+ @action(EventType.InputEvent)
+ @staticmethod
+ async def process_input(event: Event, ctx: RunnerContext) -> None:
+ user_id = InputEvent.from_event(event).input
+ future = ctx.durable_execute_async(
+ profile_service.get_profile,
+ user_id,
+ durable_id=f"profile:{user_id}",
+ )
+ profile = await future
+ ctx.send_event(OutputEvent(output=profile))
```
{{< /tab >}}
-
{{< tab "Java" >}}
-Use `ctx.durableExecuteAsync(DurableCallable)` and await the returned future.
-The optional `reconciler()` hook can be used for recovery.
```java
-@Action(EventType.InputEvent)
-public static void processInput(Event event, RunnerContext ctx) throws Exception {
- InputEvent inputEvent = InputEvent.fromEvent(event);
- DurableCallable<String> call = new DurableCallable<>() {
- @Override
- public String getId() {
- return "slow_external_call";
- }
+public class ProfileAgent extends Agent {
+ @Action(EventType.InputEvent)
+ public static void processInput(Event event, RunnerContext ctx) throws Exception {
+ String userId = (String) InputEvent.fromEvent(event).getInput();
+ DurableFuture<String> future = ctx.durableExecuteAsync(new ProfileLookup(userId));
+ String profile = future.await();
+ ctx.sendEvent(new OutputEvent(profile));
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
- @Override
- public Class<String> getResultClass() {
- return String.class;
- }
+As with ordinary async calls, creating a future does not start work; await it directly
+or include it in an awaited `ctx.gather(...)` batch. A future that is never awaited does
+not execute or save a result. A batch may mix ordinary and durable calls; only the durable
+calls persist their outcomes for recovery. The optional Python `reconciler` argument and
+Java `reconciler()` method work with durable async calls too.
+
+### Execution Requirements
+
+#### JDK and Scheduling Requirements
+
+There are two forms of concurrency: running several calls within one Action using
+`gather`, and letting other Actions progress while one Action waits. Concurrent batches
+are supported in Java and Python on all supported JDK versions, within the configured
+worker and batch limits. Concurrency between Actions has additional requirements:
+
+- **Java on JDK 21+:** add `--add-exports=java.base/jdk.internal.vm=ALL-UNNAMED` to the JVM
+ options, including `env.java.opts.all` on a cluster, to enable continuation-based execution.
+- **Java on JDK 11–20:** the experimental parallel execution engine is enabled by default
+ for plans whose Actions are all Java. Setting `parallel-execution.enabled` to `false`,
+ or including Python Actions, disables it. A Java single-call await then runs synchronously;
+ use `gather` for concurrent calls within an Action. Plans using Python Resources also need
+ the cross-language Flink versions below.
+- **Python Actions:** use `async def` and the framework's futures and `gather` APIs.
+
+See [Core Options]({{< ref "docs/operations/configuration#core-options" >}}) for worker-pool
+and concurrency limits. These scheduling requirements apply to both ordinary and durable
+async calls. Durable calls additionally require the configured action state store.
+
+#### Cross-language Async Requirements
+
+Async calls between Java and Python require Flink **1.20.5+**, **2.0.2+**, **2.1.3+**,
+**2.2.1+**, or **2.3+**, within the respective release line. This includes built-in chat
+and context-retrieval Actions that use a Resource implemented in the other language.
+
+A call may cross into the other language and call back once into its starting language.
+For example, a Python Action can call a Java chat model setup that uses a Python connection.
+The connection then returns its result through the existing call chain.
+
+| Call chain within one async operation | Usage |
+| --- | --- |
+| Python Action → Java Resource | Supported. |
+| Java Action → Python Resource | Supported. |
+| Python Action → Java chat model setup → Python connection | Supported: one callback into the starting language. |
+| Java Action → Python chat model setup → Java connection | Supported: one callback into the starting language. |
+| Python → Java → Python → Java, or Java → Python → Java → Python | Avoid: further nested calls can deadlock. |
+
+The last row describes a new call made before the earlier calls return, not a normal
+return path. For example, a Python connection called by a Java setup should not make
+another nested Resource call into Java. Keep the remaining work in that connection's
+language, or return to the Action and have it send an Event to trigger a later step. See
+[Choose APIs and Languages]({{< ref "docs/development/create-agent/choose-apis-and-languages" >}})
+for dependency and payload requirements.
+
+## Appendix
- @Override
- public String call() throws Exception {
- Thread.sleep(2000);
- return "Processed: " + inputEvent.getInput();
- }
- };
+### Trigger Condition Reference
- DurableFuture<String> future = ctx.durableExecuteAsync(call);
- String result = future.await();
- ctx.sendEvent(new OutputEvent(result));
-}
+#### Custom Event-Type Names
+
+A custom Event type may be a bare name such as `order.created` or `order-created`. Each
+dot-separated segment must start with an ASCII letter or underscore and may then contain
+ASCII letters, digits, underscores, or hyphens.
+
+A condition containing only a name, such as `urgent`, matches that Event type. To check
+an attribute instead, write a Boolean expression such as `attributes.urgent == true`.
+
+Quote a name that contains other punctuation or would otherwise be parsed as an expression,
+for example `'order:created'`, `'true'`, or `'EventType.custom'`. The quotes are part of the
+condition string: use `@action("'order:created'")` in Python or `@Action("'order:created'")`
+in Java. A quoted name matches the literal Event-type string. Quoted names must be non-empty
+and cannot contain whitespace, quotes, backslashes, or control characters.
+
+#### Expression Variables
+
+Java and Python Actions support the same condition expressions. Expressions can access:
+
+| Variable | Value |
+| --- | --- |
+| `type` | The Event type string |
+| `id` | The Event ID as a string |
+| `EventType` | The built-in Event-type constants |
+| `attributes` | The Event's top-level attribute map |
+
+For example, `type == EventType.InputEvent` checks whether the Event is an `InputEvent`.
+You can combine this comparison with attribute checks using `&&`.
+
+#### Accessing Attributes
+
+- Access an attribute as `attributes.score` or by its bare name, `score`.
+- Use `attributes["type"]` or `attributes["id"]` for attributes whose names overlap with
+ framework variables.
+- Use a literal index for a key containing dots, such as `attributes["a.b.c"]`. Check its
+ presence with `"a.b.c" in attributes`.
+
+#### Missing Attributes and Null {#missing-attributes-and-value-types}
+
+A missing attribute and a present attribute with a `null` value are different.
+`has(attributes.score)` checks whether the attribute exists; it does not exclude `null`.
+To test an optional, nullable score, use:
+
+```text
+has(attributes.score) && attributes.score != null && attributes.score > 80
```
-{{< /tab >}}
-{{< /tabs >}}
+The guard lets Events without a score fail to match normally, without trying to read a
+missing value.
+
+#### Condition Evaluation Failures
+
+Reading a missing attribute without a guard or applying an operation to an incompatible
+value can fail while evaluating a condition for an Event. For failure handling options, see
+[`action.trigger-condition.evaluate-failure-strategy`]({{< ref "docs/operations/configuration#core-options" >}})
+in Configuration.
+
+#### Limitations
+
+- `has(...)` is the only supported CEL macro. The comprehension macros `exists`,
+ `exists_one`, `all`, `filter`, and `map` are not supported.
+- Select a top-level attribute by a fixed name. Dynamic root access such as `attributes[key]`
+ and operations on the whole `attributes` map are unsupported. Dynamic access within a
+ selected attribute, such as `attributes.details[key]`, is supported.
diff --git a/docs/content/docs/development/create-agent/choose-apis-and-languages.md b/docs/content/docs/development/create-agent/choose-apis-and-languages.md
index 21a4e228..d6fc0f33 100644
--- a/docs/content/docs/development/create-agent/choose-apis-and-languages.md
+++ b/docs/content/docs/development/create-agent/choose-apis-and-languages.md
@@ -104,7 +104,9 @@ Both APIs provide asynchronous execution and `gather` for running multiple calls
Python Actions use `async def` and `await`. Java Actions use
`executeAsync(...).await()`, and Action scheduling and required JVM options depend on
the JDK version. See [Async Execution]({{< ref "docs/development/create-agent/actions#async-execution" >}})
-for the complete API and runtime requirements.
+for call and batch examples, and
+[Execution Requirements]({{< ref "docs/development/create-agent/actions#execution-requirements" >}})
+for runtime prerequisites.
### Available integrations by language
@@ -185,6 +187,12 @@ Follow these rules:
- Keep integers within the target type's range and use finite floating-point values.
Encode exact decimals and larger numbers as strings when precision must be preserved.
+These conversion rules describe application payloads. Built-in Event classes and their
+declared typed attributes are restored automatically at the JSON boundary; custom Event
+subclasses require explicit reconstruction. See
+[Serialization and Typed Values]({{< ref "docs/development/create-agent/events#json-serialization" >}})
+for examples of restoring Event and payload types.
+
{{< hint info >}}
Flink Agents uses a Java runtime. Data from the Python API crosses the Python-Java
boundary even when every Action and Resource is implemented in Python. The rules in
diff --git a/docs/content/docs/development/create-agent/events.md b/docs/content/docs/development/create-agent/events.md
index 9388b1c4..72ae30b8 100644
--- a/docs/content/docs/development/create-agent/events.md
+++ b/docs/content/docs/development/create-agent/events.md
@@ -22,201 +22,257 @@ specific language governing permissions and limitations
under the License.
-->
-## Event
+Events carry data between Actions. Each Event has:
-Events are JSON-serializable messages passed between actions. Every event has a `type` string and an
-`attributes` map that carries its payload. Action routing can match the exact event type or evaluate
-a condition expression against the event data. One event may trigger multiple actions.
+| Field | Purpose |
+| --- | --- |
+| `type` | Identifies the Event type used to match Actions. |
+| `attributes` | Holds the Event's payload as named values. |
+| `id` | Identifies the Event. A new ID is generated by default when the Event is created and cannot be changed afterward. |
-### Special Events
+One Event can trigger multiple Actions. For matching by type or payload, see
+[Trigger Conditions]({{< ref "docs/development/create-agent/actions#trigger-conditions" >}}).
-* `InputEvent`: Generated by the framework when an input record arrives. The record is available in
- the event's `input` attribute. Actions whose trigger conditions match an `InputEvent` are the
- agent's entry points.
-* `OutputEvent`: When an action sends this event, the framework emits its `output` attribute
- downstream as an agent output. It bypasses action matching.
+## Input and Output Events {#special-events}
-### Unified Event
+`InputEvent` and `OutputEvent` connect an Agent to the surrounding Flink pipeline:
-For simple cases, users can pass data between actions directly using `Event` with a custom `type` and `attributes`, without needing to define a subclass. For more structured events, see [Custom Event Subclasses](#custom-event-subclasses) below.
+| Event | Role | Payload |
+| --- | --- | --- |
+| `InputEvent` | The framework creates one for each input record. Matching Actions start processing that record. | `input` contains the record. |
+| `OutputEvent` | An Action sends one to emit a result downstream. It bypasses Action matching. | `output` contains the result. |
-{{< tabs "Unified Event" >}}
+Use `InputEvent.from_event(event).input` in Python or
+`InputEvent.fromEvent(event).getInput()` in Java to read an input record.
+## Creating and Handling Events {#unified-event}
+
+Create an `Event` with a custom type and attributes to pass data to another Action.
+A subclass is optional. The following snippets use the `review_ready` Event from
+[Sending Events]({{< ref "docs/development/create-agent/actions#sending-events" >}}).
+They run inside Action methods; framework imports and job setup are omitted.
+
+{{< tabs "Creating and Handling Events" >}}
{{< tab "Python" >}}
+Send the cleaned review from the first Action:
+
```python
-# Send a unified event from one action
-@action(EventType.InputEvent)
-@staticmethod
-def create_my_event(event: Event, ctx: RunnerContext) -> None:
- ctx.send_event(
- Event(type="my_event", attributes={"field1": "test", "field2": 42})
- )
-
-# Consume it in another action
-@action("my_event")
-@staticmethod
-def handle_my_event(event: Event, ctx: RunnerContext) -> None:
- field1: str = event.get_attr("field1")
- field2: int = event.get_attr("field2")
+review = InputEvent.from_event(event).input
+ctx.send_event(Event(type="review_ready", attributes={"text": review.strip()}))
```
-{{< /tab >}}
-{{< tab "Java" >}}
-```java
-// Send a unified event from one action
-@Action(EventType.InputEvent)
-public static void createMyEvent(Event event, RunnerContext ctx) {
- ctx.sendEvent(new Event("my_event", Map.of("field1", "test", "field2", 42)));
-}
+In an Action registered with `@action("review_ready")`, read the payload:
-// Consume it in another action
-@Action("my_event")
-public static void handleMyEvent(Event event, RunnerContext ctx) {
- String field1 = (String) event.getAttr("field1");
- int field2 = (int) event.getAttr("field2");
-}
+```python
+text = event.get_attr("text")
+ctx.send_event(OutputEvent(output=text))
```
{{< /tab >}}
+{{< tab "Java" >}}
+Send the cleaned review from the first Action:
-{{< /tabs >}}
-
-{{< hint info >}}
-The runtime records Event IDs and source relationships in [Trace Log]({{< ref "docs/operations/monitoring#json-format" >}}), separately from the Event payload. Source relationships are not fields of the Event API, and callers do not need to manage them when sending Events. Keep user data in `attributes`.
-{{< /hint >}}
-
-### JSON Serialization
-
-Events are serialized as JSON when passed between Python actions or across the Java-Python boundary. When an event crosses that boundary, its **type** and its **payload** are handled differently.
-
-**Built-in event types are restored automatically.** A dispatched `InputEvent`,
-`ChatRequestEvent`, `ChatResponseEvent`, or other built-in Event has its concrete
-type and validated attributes. Use it directly inside an Action, with a cast in Java.
-
-**Custom event types stay generic.** A user-defined event type is not known to the framework, so it arrives as a generic `Event`; call your subclass's `from_event` / `fromEvent` to reconstruct it (see [Custom Event Subclasses](#custom-event-subclasses)).
+```java
+String review = (String) InputEvent.fromEvent(event).getInput();
+ctx.sendEvent(new Event("review_ready", Map.of("text", review.trim())));
+```
-**Arbitrary payload values are not reconstructed.** Restoration recovers the event's type and its typed built-in attributes (for example, `ChatRequestEvent.messages` become typed `ChatMessage` objects), but it does not rebuild arbitrary payload values. Non-trivial values nested in `attributes` — such as Pydantic models, or the record carried by `InputEvent.input` — lose their type information and arrive as plain `dict` / `Map` objects, so you must reconstruct them yourself:
+In an Action registered with `@Action("review_ready")`, read the payload:
-```python
-# `event` is already a concrete InputEvent (auto-restored at the JSON boundary);
-# only its opaque `input` payload needs manual reconstruction.
-input_data = ItemData.model_validate(event.input)
+```java
+String text = (String) event.getAttr("text");
+ctx.sendEvent(new OutputEvent(text));
```
+{{< /tab >}}
+{{< /tabs >}}
-For supported payload values, see
-[Data passed between Java and Python]({{< ref "docs/development/create-agent/choose-apis-and-languages#data-passed-between-java-and-python" >}}).
+## Custom Event Subclasses
-### Custom Event Subclasses
+Define a subclass when you want to reuse an Event type with named accessors and payload
+validation. Keep its payload in `attributes`. The `ReviewReadyEvent` below represents
+the same `review_ready` type and `text` payload used above.
-Users can also define custom event subclasses for reusable, structured events. Data should be stored in the `attributes` map, and the subclass must implement a `from_event` / `fromEvent` factory method that validates required attributes and reconstructs typed objects from the deserialized data.
+When the framework deserializes an Event at a JSON boundary, custom types arrive as
+generic `Event` objects. Provide `from_event` / `fromEvent` to validate the payload and
+obtain your typed representation.
+Use `reconstruct_from` / `reconstructFrom` to preserve the original Event's ID and
+attachments.
{{< tabs "Custom Event" >}}
-
{{< tab "Python" >}}
```python
-class MyEvent(Event):
- EVENT_TYPE: ClassVar[str] = "my_event"
+from typing import ClassVar
+
+from flink_agents.api.events.event import Event
+
- def __init__(self, value: str) -> None:
- super().__init__(type=MyEvent.EVENT_TYPE, attributes={"value": value})
+class ReviewReadyEvent(Event):
+ EVENT_TYPE: ClassVar[str] = "review_ready"
+
+ def __init__(self, text: str) -> None:
+ if not isinstance(text, str):
+ raise ValueError("ReviewReadyEvent requires a string 'text' attribute")
+ super().__init__(type=self.EVENT_TYPE, attributes={"text": text})
@classmethod
- @override
- def from_event(cls, event: Event) -> "MyEvent":
- assert "value" in event.attributes
- result = cls(value=event.attributes["value"])
+ def from_event(cls, event: Event) -> "ReviewReadyEvent":
+ if event.type != cls.EVENT_TYPE:
+ raise ValueError("Expected a review_ready Event")
+ result = cls(text=event.get_attr("text"))
return result.reconstruct_from(event)
@property
- def value(self) -> str:
- return self.get_attr("value")
+ def text(self) -> str:
+ return self.get_attr("text")
```
{{< /tab >}}
-
{{< tab "Java" >}}
```java
-public class MyEvent extends Event {
- public static final String EVENT_TYPE = "my_event";
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.agents.api.Event;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.UUID;
+
+public class ReviewReadyEvent extends Event {
+ public static final String EVENT_TYPE = "review_ready";
- public MyEvent(String value) {
- super(EVENT_TYPE);
- setAttr("value", value);
+ public ReviewReadyEvent(String text) {
+ this(null, Collections.singletonMap("text", text));
}
@JsonCreator
- public MyEvent(
+ public ReviewReadyEvent(
@JsonProperty("id") UUID id,
@JsonProperty("attributes") Map<String, Object> attributes) {
super(id, EVENT_TYPE, attributes);
+ if (!(getAttr("text") instanceof String)) {
+ throw new IllegalArgumentException(
+ "ReviewReadyEvent requires a string 'text' attribute");
+ }
}
- public static MyEvent fromEvent(Event event) {
- return reconstructFrom(event, MyEvent::new);
+ public static ReviewReadyEvent fromEvent(Event event) {
+ if (!EVENT_TYPE.equals(event.getType())) {
+ throw new IllegalArgumentException("Expected a review_ready Event");
+ }
+ return reconstructFrom(event, ReviewReadyEvent::new);
}
- public String getValue() {
- return (String) getAttr("value");
+ @JsonIgnore
+ public String getText() {
+ return (String) getAttr("text");
}
}
```
{{< /tab >}}
+{{< /tabs >}}
+
+Constructing a `ReviewReadyEvent` from text creates a new Event. `from_event` / `fromEvent`
+produces a typed representation of an existing Event, retaining its ID instead of creating
+a new Event occurrence. Both examples reject a missing or non-string `text` attribute.
+
+For Java, `@JsonCreator` and `@JsonProperty` let Jackson restore the subclass during
+recovery. The constructor accepts the saved ID and attributes, and is also used by
+`reconstructFrom`. `@JsonIgnore` on `getText()` keeps the payload in `attributes` without
+serializing a second top-level `text` property. If your payload contains custom objects,
+restore their types in this constructor too.
+
+Use the subclass when sending an Event, and its factory when handling one. These
+snippets run inside the sending and receiving Actions respectively:
+
+{{< tabs "Using a Custom Event" >}}
+{{< tab "Python" >}}
+```python
+# In the sending Action:
+ctx.send_event(ReviewReadyEvent(text="A useful review."))
+```
+
+```python
+# In an Action registered with @action(ReviewReadyEvent.EVENT_TYPE):
+review = ReviewReadyEvent.from_event(event)
+ctx.send_event(OutputEvent(output=review.text))
+```
+{{< /tab >}}
+{{< tab "Java" >}}
+```java
+// In the sending Action:
+ctx.sendEvent(new ReviewReadyEvent("A useful review."));
+```
+```java
+// In an Action registered with @Action(ReviewReadyEvent.EVENT_TYPE):
+ReviewReadyEvent review = ReviewReadyEvent.fromEvent(event);
+ctx.sendEvent(new OutputEvent(review.getText()));
+```
+{{< /tab >}}
{{< /tabs >}}
-{{< hint warning >}}
-Python Event IDs are immutable. Custom `from_event` implementations written against earlier
-versions may assign `result.id = event.id`; that assignment now raises a Pydantic
-`ValidationError`. Replace `result.id = event.id` followed by `return result` with
-`return result.reconstruct_from(event)`.
+## Serialization and Typed Values {#json-serialization}
-Typed reconstruction represents the same Event occurrence, so it preserves the original ID and
-attachments rather than creating a new occurrence:
+Event attributes must support JSON serialization. For supported values and conversions,
+see [Data passed between Java and Python]({{< ref "docs/development/create-agent/choose-apis-and-languages#data-passed-between-java-and-python" >}}).
-- **Python**: return `reconstruct_from(event)` after constructing the typed object.
- It returns a new typed object with the Event's UUIDv4 `id` and a copy of its
- attachments; the returned Event's `id` remains immutable.
-- **Java**: implement the typed constructor that accepts `(UUID id, Map<String, Object>
- attributes)`, then have `fromEvent` return
- `reconstructFrom(event, MyEvent::new)`. The framework supplies the original ID and
- attributes, validates that the ID is preserved, and carries over attachments and
- `sourceTimestamp`.
-{{< /hint >}}
+### Event Transfer and Replay
-{{< hint info >}}
-All attribute values must be JSON-serializable. In Python, this means `BaseModel`-serializable or primitive types. In Java, values must be Jackson-serializable.
-{{< /hint >}}
+Whether you need to restore payload types depends on how the Event reaches an Action:
-{{< hint warning >}}
-When defining a custom `Event` subclass in Java, annotate its `(UUID id, Map<String, Object> attributes)` constructor with `@JsonCreator`, and annotate both parameters with `@JsonProperty`. Durable execution stores events in `ActionState` and uses Jackson to restore concrete event subclasses during recovery. Without these annotations, recovery deserialization fails because the base `Event` constructor's `@JsonCreator` is not inherited by subclasses.
+- **Between Java Actions:** normal execution passes Java objects directly, preserving
+ their types. No payload reconstruction is needed.
+- **To or from Python Actions:** Events pass through JSON serialization. Flink runs on
+ a Java runtime, so this also applies between two Python Actions.
+- **Replay from an ActionStateStore:** persisted Events are restored from JSON.
+ Application-defined payload objects may become `Map` values even in a Java-only Agent.
-If the subclass stores typed objects in `attributes`, convert them back to their typed forms inside this constructor too, as the built-in events do. For example, `ChatResponseEvent` restores its typed attributes like this:
+### Restoring Typed Values
-```java
-@JsonCreator
-public ChatResponseEvent(
- @JsonProperty("id") UUID id,
- @JsonProperty("attributes") Map<String, Object> attributes) {
- super(id, EVENT_TYPE, normalizeAttributes(attributes));
-}
+At the Python–Java boundary, restoring an Event's type and restoring objects inside
+its payload are separate steps:
-/** Converts nested attributes back to their typed forms. */
-private static Map<String, Object> normalizeAttributes(Map<String, Object> attributes) {
- Object rawId = attributes.get("request_id");
- if (rawId instanceof String) {
- attributes.put("request_id", UUID.fromString((String) rawId));
- }
- Object rawResponse = attributes.get("response");
- if (rawResponse instanceof Map) {
- attributes.put("response", MAPPER.convertValue(rawResponse, ChatResult.class));
- }
- return attributes;
-}
+| Data at a JSON boundary | What the receiving Action gets |
+| --- | --- |
+| Built-in Event, such as `InputEvent` or `ChatRequestEvent` | The corresponding built-in Event class. Framework-declared field types are also restored; for example, `ChatRequestEvent.messages` contains `ChatMessage` objects. |
+| Custom Event, such as `ReviewReadyEvent` | A generic `Event`. Call your subclass's `from_event` / `fromEvent` for typed access. |
+| Application-defined object in the payload, such as a review model in `InputEvent.input` or `ChatResponseEvent.structured_output` | JSON data, typically a `dict` / `Map`. Reconstruct the application type if needed. |
+
+Restoring a built-in Event does not restore application-defined types inside it. For
+example, a restored `ChatResponseEvent` can still carry structured output as a `dict`
+or `Map` rather than your application class.
+
+For example, if your application defines a `ReviewData` type, convert an input payload
+inside the receiving Action as follows. `ReviewData` is a Pydantic model in Python or
+a Jackson-deserializable class in Java:
+
+{{< tabs "Restore an Input Payload" >}}
+{{< tab "Python" >}}
+```python
+review = ReviewData.model_validate(InputEvent.from_event(event).input)
+```
+{{< /tab >}}
+{{< tab "Java" >}}
+```java
+ReviewData review = new ObjectMapper().convertValue(
+ InputEvent.fromEvent(event).getInput(), ReviewData.class);
```
-{{< /hint >}}
+
+`ObjectMapper` is imported from `com.fasterxml.jackson.databind`.
+{{< /tab >}}
+{{< /tabs >}}
## Built-in Events and Actions
-There are several built-in `Event` and `Action` in Flink-Agents:
-* See [Chat Flow]({{< ref "docs/development/calling-llms/chat#chat-events" >}}) for how to chat with a LLM leveraging built-in action and events.
-* See [Tool Use]({{< ref "docs/development/calling-llms/tool_use#built-in-events-and-actions" >}}) for how to programmatically use a tool leveraging built-in action and events.
-* See [Vector Stores]({{< ref "docs/development/vector-search/vector_stores#built-in-events-and-actions" >}}) for how to retrieve context from vector stores leveraging built-in action and events.
-* See [Model Routing]({{< ref "docs/development/calling-llms/model_routing#observability" >}}) for how a chat request selects between several models and how routing decisions are reported.
+Flink Agents provides built-in Actions for common operations such as calling chat
+models, executing tools, and retrieving context. These Actions are registered with
+your Agent by default, so you do not need to implement or register them yourself.
+
+Register the Resources required by the operation, then send the corresponding request
+Event from your Action. The built-in Action handles the request and sends a response
+Event that your Actions can handle to continue the workflow:
+
+| Task | Events | Guide |
+| --- | --- | --- |
+| Call a chat model | `ChatRequestEvent`, `ChatResponseEvent` | [Chat with Models]({{< ref "docs/development/calling-llms/chat#chat-events" >}}) |
+| Call tools | `ToolRequestEvent`, `ToolResponseEvent` | [Tool Use]({{< ref "docs/development/calling-llms/tool_use#built-in-events-and-actions" >}}) |
+| Retrieve context | `ContextRetrievalRequestEvent`, `ContextRetrievalResponseEvent` | [Vector Stores]({{< ref "docs/development/vector-search/vector_stores#built-in-events-and-actions" >}}) |
diff --git a/docs/content/docs/development/create-agent/overview.md b/docs/content/docs/development/create-agent/overview.md
index aef72747..6301579c 100644
--- a/docs/content/docs/development/create-agent/overview.md
+++ b/docs/content/docs/development/create-agent/overview.md
@@ -24,206 +24,162 @@ specific language governing permissions and limitations
under the License.
-->
-## Overview
+## What Is an Agent?
-An Agent in Flink-Agents is an agent whose reasoning and behavior are organized as a directed workflow of modular steps, called actions, connected by events. This design is inspired by the need to orchestrate complex, multi-stage tasks in a transparent, extensible, and data-centric way, leveraging Apache Flink's streaming architecture.
+An AI agent uses a language model to reason about a task, decide what to do next, and
+take actions toward a goal. It can use tools to interact with external systems and
+memory to retain context, adapting its next steps based on the results.
-In Flink-Agents, an Agent is a class that inherits from the `Agent` base class. Its workflow
-is defined by actions: functions decorated with `@action(...)` in Python or methods annotated with
-`@Action(...)` in Java. Each action declares one or more trigger conditions, which may be exact event
-types or condition expressions. Actions consume events, perform reasoning or tool calls, and emit new
-events that may trigger other actions. This event-driven workflow forms a directed graph that may
-contain cycles. Each node is an action, and each edge represents an event flow between actions.
+Flink Agents provides an event-driven model for building and orchestrating these agents
+on Apache Flink. **Actions** define the work to perform, **Events** carry data and connect
+steps, and **Resources** provide reusable capabilities such as models, tools, and prompts.
+You can explicitly define the workflow, let a model choose the next steps within it,
+or combine both approaches.
-An Agent is well-suited for scenarios where the solution requires explicit orchestration, branching, or multi-step reasoning, such as data enrichment, multi-tool pipelines, or complex business logic.
+## Core Concepts
-## Agent Example
+| Concept | Role in an Agent | Example |
+| --- | --- | --- |
+| **Event** | Carries data between Actions. Its type and attributes determine which Actions react to it. | A `ChatRequestEvent` carries messages and the name of a model to call. |
+| **Action** | Reacts to matching Events, performs work, and can send new Events. One Event can trigger multiple Actions. | An Action reads a review and sends a request for a summary. |
+| **Resource** | A reusable, named runtime component that Actions or other Resources can use. | A chat model configured for summarizing reviews. |
-{{< tabs "Agent Example" >}}
+Common Resources include:
-{{< tab "Python" >}}
-```python
-class ReviewAnalysisAgent(Agent):
- """An agent that uses a large language model (LLM) to analyze product reviews
- and generate a satisfaction score and potential reasons for dissatisfaction.
+- [Chat Models]({{< ref "docs/development/calling-llms/chat-models" >}}) for model calls.
+- [Tools]({{< ref "docs/development/calling-llms/tool_use" >}}) for callable operations.
+- [Prompts]({{< ref "docs/development/calling-llms/prompts" >}}) for reusable message templates.
+- [MCP Servers]({{< ref "docs/development/calling-llms/mcp" >}}) for accessing tools and prompts provided by external MCP servers.
+- [Vector Stores]({{< ref "docs/development/vector-search/vector_stores" >}}) for storing and retrieving embeddings and related data.
+- [Skills]({{< ref "docs/development/calling-llms/skills" >}}) for reusable instructions and supporting files.
- This agent receives a product review and produces a satisfaction score and a list
- of reasons for dissatisfaction. It handles prompt construction, LLM interaction,
- and output parsing.
- """
+[Memory]({{< ref "docs/development/memory/overview" >}}) lets Actions retain and retrieve
+information through the execution context. It is a separate API, not a Resource type.
- @prompt
- @staticmethod
- def review_analysis_prompt() -> Prompt:
- """Prompt for review analysis."""
- return review_analysis_prompt
+## How an Agent Runs
- @tool
- @staticmethod
- def notify_shipping_manager(id: str, review: str) -> None:
- """Notify the shipping manager when product received a negative review due to
- shipping damage.
-
- Parameters
- ----------
- id : str
- The id of the product that received a negative review due to shipping damage
- review: str
- The negative review content
- """
- # reuse the declared function, but for parsing the tool metadata, we write doc
- # string here again.
- notify_shipping_manager(id=id, review=review)
+Consider an Agent that summarizes product reviews. You define two Actions: one to prepare
+the request and one to emit the result. The framework supplies the chat Action between them.
+
+{{< mermaid >}}
+flowchart TD
+ input[Input review] -->|InputEvent| request[Your Action: request summary]
+ request -->|ChatRequestEvent| chat[Built-in chat Action]
+ chat -. uses .-> model[Resource: review model]
+ chat -->|ChatResponseEvent| output[Your Action: emit summary]
+ output -->|OutputEvent| downstream[Downstream output]
+{{< /mermaid >}}
+
+1. The framework wraps each input record in an `InputEvent`.
+2. Your input Action reacts to that Event and sends a `ChatRequestEvent`, referencing
+ the registered model by name.
+3. The built-in chat Action calls the model and sends a `ChatResponseEvent` with the result.
+4. Your response Action sends an `OutputEvent`. The framework emits its payload downstream;
+ `OutputEvent` bypasses Action matching.
+
+Actions are connected by the Events they send and react to, rather than by direct calls to
+one another. The example is a linear flow, but an Event can trigger several matching Actions,
+and an Action can send Events that lead to different branches or another iteration of a loop.
+
+## Define Your Agent
+
+To define this workflow:
+
+1. Extend `Agent` and declare the Resources it needs.
+2. Register Actions with `@action(...)` in Python or `@Action(...)` in Java, specifying
+ the Events they react to.
+3. Use the `RunnerContext` passed to each Action to send Events and access runtime services.
+
+### Registering and Referencing Resources
+
+Register a Resource with a name and a resource type. For example, `@chat_model_setup`
+in Python or `@ChatModelSetup` in Java registers a chat model under the declaration's method
+name. You can also use `add_resource` / `addResource` to supply the name and type explicitly.
+Register a Resource on the agents execution environment to make it available to multiple Agents.
+
+Inside an Action, use the execution context to get a Resource instance by its name and type,
+then use the instance's methods. For the model declared in the example below:
+
+- **Python:** `model = ctx.get_resource("review_model", ResourceType.CHAT_MODEL)`
+- **Java:** `Resource model = ctx.getResource("reviewModel", ResourceType.CHAT_MODEL)`
+
+Built-in Actions handle some Resource operations for you. For example, when you send a
+`ChatRequestEvent`, the [built-in chat Action]({{< ref "docs/development/calling-llms/chat#chat-events" >}})
+gets the named chat model Resource from the
+context, calls it, and sends a `ChatResponseEvent`. The example below uses this built-in
+Action, so your own Actions do not need to get and call the model directly. Sending an Event
+works here because an Action handles that Event and performs the Resource operation.
+
+Here, `review_model` / `reviewModel` names the Resource within your application;
+`qwen3:8b` selects the model on the Ollama server.
+
+### Review Summary Example
+The examples below accept a review as a **string** and output the model's summary as a string.
+They assume an Ollama connection named `ollama_server` is registered in the agents execution
+environment and the `qwen3:8b` model is available on that server. See the
+[Quickstart]({{< ref "docs/get-started/quickstart/create-agent/single-agent#prepare-agents-execution-environment" >}})
+for environment registration (use `ollama_server` as the connection name in either language),
+and [Chat Model Providers]({{< ref "docs/development/calling-llms/chat-models/providers" >}})
+for provider configuration. Imports and job setup are omitted here to focus on the Agent definition.
+
+{{< tabs "Define Your Agent" >}}
+
+{{< tab "Python" >}}
+```python
+class ReviewSummaryAgent(Agent):
@chat_model_setup
@staticmethod
- def review_analysis_model() -> ResourceDescriptor:
- """ChatModel which focus on review analysis."""
+ def review_model() -> ResourceDescriptor:
return ResourceDescriptor(
clazz=ResourceName.ChatModel.OLLAMA_SETUP,
connection="ollama_server",
model="qwen3:8b",
- prompt="review_analysis_prompt",
- tools=["notify_shipping_manager"],
- extract_reasoning=True,
)
@action(EventType.InputEvent)
@staticmethod
- def process_input(event: Event, ctx: RunnerContext) -> None:
- """Process input event and send chat request for review analysis."""
- input_event = InputEvent.from_event(event)
- input: ProductReview = input_event.input
- ctx.short_term_memory.set("id", input.id)
-
- content = f"""
- "id": {input.id},
- "review": {input.review}
- """
- msg = ChatMessage(role=MessageRole.USER)
+ def request_summary(event: Event, ctx: RunnerContext) -> None:
+ review = InputEvent.from_event(event).input
ctx.send_event(
ChatRequestEvent(
- model="review_analysis_model",
- messages=[msg],
- prompt_args={"input": content},
+ model="review_model",
+ messages=[ChatMessage.user(f"Summarize this review: {review}")],
)
)
@action(EventType.ChatResponseEvent)
@staticmethod
- def process_chat_response(event: Event, ctx: RunnerContext) -> None:
- """Process chat response event and send output event."""
- chat_response = ChatResponseEvent.from_event(event)
- try:
- json_content = json.loads(chat_response.response.text)
- ctx.send_event(
- OutputEvent(
- output=ProductReviewAnalysisRes(
- id=ctx.short_term_memory.get("id"),
- score=json_content["score"],
- reasons=json_content["reasons"],
- )
- )
- )
- except Exception:
- logging.exception(
- f"Error processing chat response {chat_response.response.text}"
- )
-
- # To fail the agent, you can raise an exception here.
+ def emit_summary(event: Event, ctx: RunnerContext) -> None:
+ response = ChatResponseEvent.from_event(event).response
+ ctx.send_event(OutputEvent(output=response.text))
```
{{< /tab >}}
{{< tab "Java" >}}
```java
-/**
- * An agent that uses a large language model (LLM) to analyze product reviews and generate a
- * satisfaction score and potential reasons for dissatisfaction.
- *
- * <p>This agent receives a product review and produces a satisfaction score and a list of reasons
- * for dissatisfaction. It handles prompt construction, LLM interaction, and output parsing.
- */
-public class ReviewAnalysisAgent extends Agent {
-
- private static final ObjectMapper MAPPER = new ObjectMapper();
-
- @Prompt
- public static org.apache.flink.agents.api.prompt.Prompt reviewAnalysisPrompt() {
- return REVIEW_ANALYSIS_PROMPT;
- }
-
+public class ReviewSummaryAgent extends Agent {
@ChatModelSetup
- public static ResourceDescriptor reviewAnalysisModel() {
+ public static ResourceDescriptor reviewModel() {
return ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
- .addInitialArgument("connection", "ollamaChatModelConnection")
+ .addInitialArgument("connection", "ollama_server")
.addInitialArgument("model", "qwen3:8b")
- .addInitialArgument("prompt", "reviewAnalysisPrompt")
- .addInitialArgument("tools", Collections.singletonList("notifyShippingManager"))
- .addInitialArgument("extract_reasoning", true)
.build();
}
- /**
- * Tool for notifying the shipping manager when product received a negative review due to
- * shipping damage.
- *
- * @param id The id of the product that received a negative review due to shipping damage
- * @param review The negative review content
- */
- @Tool(
- description =
- "Notify the shipping manager when product received a negative review due to shipping damage.")
- public static void notifyShippingManager(
- @ToolParam(name = "id") String id, @ToolParam(name = "review") String review) {
- CustomTypesAndResources.notifyShippingManager(id, review);
- }
-
- /** Process input event and send chat request for review analysis. */
@Action(EventType.InputEvent)
- public static void processInput(Event event, RunnerContext ctx) throws Exception {
- InputEvent inputEvent = InputEvent.fromEvent(event);
- String input = (String) inputEvent.getInput();
- MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
- CustomTypesAndResources.ProductReview inputObj =
- MAPPER.readValue(input, CustomTypesAndResources.ProductReview.class);
-
- ctx.getShortTermMemory().set("id", inputObj.getId());
-
- String content =
- String.format(
- "{\n" + "\"id\": %s,\n" + "\"review\": \"%s\"\n" + "}",
- inputObj.getId(), inputObj.getReview());
- ChatMessage msg = new ChatMessage(MessageRole.USER, "");
-
+ public static void requestSummary(Event event, RunnerContext ctx) {
+ String review = (String) InputEvent.fromEvent(event).getInput();
ctx.sendEvent(
new ChatRequestEvent(
- "reviewAnalysisModel", List.of(msg), Map.of("input", content), null));
+ "reviewModel",
+ List.of(ChatMessage.user("Summarize this review: " + review))));
}
@Action(EventType.ChatResponseEvent)
- public static void processChatResponse(Event event, RunnerContext ctx)
- throws Exception {
- ChatResponseEvent chatResponse = ChatResponseEvent.fromEvent(event);
- JsonNode jsonNode = MAPPER.readTree(chatResponse.getResponse().getText());
- JsonNode scoreNode = jsonNode.findValue("score");
- JsonNode reasonsNode = jsonNode.findValue("reasons");
- if (scoreNode == null || reasonsNode == null) {
- throw new IllegalStateException(
- "Invalid response from LLM: missing 'score' or 'reasons' field.");
- }
- List<String> result = new ArrayList<>();
- if (reasonsNode.isArray()) {
- for (JsonNode node : reasonsNode) {
- result.add(node.asText());
- }
- }
-
- ctx.sendEvent(
- new OutputEvent(
- new CustomTypesAndResources.ProductReviewAnalysisRes(
- ctx.getShortTermMemory().get("id").getValue().toString(),
- scoreNode.asInt(),
- result)));
+ public static void emitSummary(Event event, RunnerContext ctx) {
+ String summary = ChatResponseEvent.fromEvent(event).getResponse().getText();
+ ctx.sendEvent(new OutputEvent(summary));
}
}
```
@@ -231,9 +187,13 @@ public class ReviewAnalysisAgent extends Agent {
{{< /tabs >}}
-Next, learn how [Actions]({{< ref "docs/development/create-agent/actions" >}})
-process incoming Events and emit new Events, and how
-[Events]({{< ref "docs/development/create-agent/events" >}}) carry data and trigger
-Actions. Then see
-[Choose APIs and Languages]({{< ref "docs/development/create-agent/choose-apis-and-languages" >}})
-to choose among the Java, Python, and YAML APIs and combine Java and Python components.
+## Next Steps
+
+- Learn how to register Actions, define trigger conditions, and use async or durable execution
+ in [Actions]({{< ref "docs/development/create-agent/actions" >}}).
+- Learn how to define and pass data in [Events]({{< ref "docs/development/create-agent/events" >}}).
+- [Choose APIs and Languages]({{< ref "docs/development/create-agent/choose-apis-and-languages" >}})
+ for your application, or define an Agent with the
+ [YAML API]({{< ref "docs/development/create-agent/yaml" >}}).
+- Follow the [Quickstart]({{< ref "docs/get-started/quickstart/create-agent/single-agent" >}})
+ to connect an Agent to a Flink job and build a complete review-analysis application.
diff --git a/docs/content/docs/development/create-agent/yaml.md b/docs/content/docs/development/create-agent/yaml.md
index 5f0fb6f8..cb18bda1 100644
--- a/docs/content/docs/development/create-agent/yaml.md
+++ b/docs/content/docs/development/create-agent/yaml.md
@@ -31,11 +31,8 @@ Beyond the Python and Java programmatic APIs, Flink Agents offers a **declarativ
- **Human-friendly**: low entry barrier and easy to templatize across deployments.
- **Coding-agent-friendly**: a fixed schema reduces token cost, enables strict schema validation, and decouples configuration changes (declared parameters) from logic changes (action code).
-A Flink Agents application is composed of three concepts; the YAML API exposes all three as declarative sections:
-
-- **Events**: the messages flowing inside an agent. Every event has a type used for routing.
-- **Actions**: code snippets triggered by events and emitting new events. They carry the agent's business logic.
-- **Resources**: reusable components — Chat Models, Tools, Prompts, Vector Stores, MCP servers, etc. — referenced by actions at runtime.
+The YAML API uses the same Events, Actions, and Resources described in
+[Core Concepts]({{< ref "docs/development/create-agent/overview#core-concepts" >}}).
A YAML file describes **Resources** and **Actions** declaratively; the **Event** types are referenced by name (built-in event aliases like `input` / `chat_response`, or your own event-type strings). The implementation classes (action functions, tool functions, custom types) still live in your Python or Java code; the YAML file only describes *how* those pieces are wired together.
@@ -109,7 +106,15 @@ agents:
#### Resources
-All resource sections are declared as lists under the agent. Each entry has a `name` unique within its section.
+Declare Resources as lists under the agent, with a `name` unique within each section.
+The section determines the resource type: for example, `chat_model_setups` registers
+`CHAT_MODEL` Resources and `tools` registers `TOOL` Resources. The optional `type: python`
+or `type: java` field selects the implementation language.
+
+These declarations register Resources for use by name, just like decorators, annotations,
+or `add_resource` / `addResource`. Actions can get their instances through
+`ctx.get_resource` / `ctx.getResource`; see
+[Registering and Referencing Resources]({{< ref "docs/development/create-agent/overview#registering-and-referencing-resources" >}}).
**Prompt** — declarative prompt template; pick exactly one of `text` or `messages`.
@@ -256,14 +261,18 @@ skills:
Remote skill archives contain trusted agent instructions and may contain executable scripts. Pin remote sources with `sha256`, or use `Skills.from_url_with_sha256(...)` (Python) / `Skills.fromUrlWithSha256(...)` (Java) in code.
-**ResourceDescriptor-backed resources** — `chat_model_connections`, `chat_model_setups`, `embedding_model_connections`, `embedding_model_setups`, `vector_stores`, `mcp_servers`. Unlike the resources above (which are constructed inline by the loader), these are described by a `ResourceDescriptor` and instantiated by the framework at runtime. They all share the same shape:
+**Resources configured by implementation class** — `chat_model_connections`,
+`chat_model_setups`, `embedding_model_connections`, `embedding_model_setups`, `vector_stores`,
+and `mcp_servers`. Declare the implementation and its configuration using the fields below.
+This corresponds to a `ResourceDescriptor` in Java or Python; the framework creates the
+Resource instance at runtime.
| Field | Required | Description |
|-------|----------|-------------|
| `name` | yes | Resource name (referenced by other resources or actions). |
| `clazz` | yes | Either a built-in [alias](#class-aliases) (e.g. `ollama`, `openai`) or a fully-qualified class path. |
| `type` | no | Implementation language: `python` or `java`. Defaults to `python` (see [Selecting the implementation language](#selecting-the-implementation-language)). |
-| *extras* | no | Any additional keys are forwarded verbatim as `ResourceDescriptor` init arguments (e.g. `base_url`, `endpoint`, `model`, `request_timeout`). |
+| *extras* | no | Additional configuration for the selected implementation (e.g. `base_url`, `endpoint`, `model`, `request_timeout`). |
```yaml
chat_model_connections:
@@ -279,7 +288,9 @@ mcp_servers:
endpoint: http://127.0.0.1:8000/mcp
```
-**Inter-resource references.** Several fields are not values but *name lookups* into another section, resolved within the agent first and then against shared resources. In the [single-agent example](#declaring-a-single-agent):
+**References between Resources.** Fields such as `connection`, `prompt`, and `tools`
+refer to registered Resource names. The Agent's own registrations take precedence over
+shared Resources from the execution environment. In the [single-agent example](#declaring-a-single-agent):
| Field | Refers to | Example |
|-------|-----------|---------|
@@ -288,12 +299,22 @@ mcp_servers:
| `chat_model_setups[].tools` | a list of `tools[]` names | `tools: [notify_shipping_manager]` |
| `embedding_model_setups[].connection` | an `embedding_model_connections[]` name | `connection: my_embedding_server` |
-The referenced name must exist in the same agent or be a shared resource at the file level; otherwise loading fails.
+The referenced Resource must be registered on the Agent or its execution environment.
+The setup's `name: review_analysis_model` identifies the Resource in your application;
+its `model: qwen3:8b` selects the model on the Ollama server.
+
+An Action can send a `ChatRequestEvent` naming `review_analysis_model` to use the
+[built-in chat Action]({{< ref "docs/development/calling-llms/chat#chat-events" >}}),
+which gets and calls the model Resource. This event-based path is provided by the built-in
+Action; registering a Resource alone does not create an Event handler for it.
#### Actions
`actions:` is a list. Each entry is either an inline action **map** or, when reusing a shared action, a bare **string** referring to the shared action's name.
+[Built-in Actions]({{< ref "docs/development/create-agent/events#built-in-events-and-actions" >}})
+are registered automatically; you do not need to list them under `actions:`.
+
Inline action (map) fields:
| Field | Required | Description |
diff --git a/docs/content/docs/development/memory/long_term_memory.md b/docs/content/docs/development/memory/long_term_memory.md
index d278af08..3f39325a 100644
--- a/docs/content/docs/development/memory/long_term_memory.md
+++ b/docs/content/docs/development/memory/long_term_memory.md
@@ -185,17 +185,12 @@ A memory set is scoped to the key of the action that obtained it. Call `get_memo
in a later action. Reusing a set would apply another key's operations to the key it was
originally obtained for, and operating on a set that carries no scope raises an error.
-Obtaining a memory set and deleting one both read the key currently in scope, which only
-the action's own thread reads consistently. Both raise wherever they run on a worker
-thread, so passing `getMemorySet` / `get_memory_set` or `deleteMemorySet` /
-`delete_memory_set` to `durableExecuteAsync` / `durable_execute_async` is unsupported: in
-Python, and in Java on JDK 21 and above, the callable runs on a worker thread and the call
-raises. On JDK 11 Java runs the callable inline instead, so the same code does not raise
-today, but it is unsupported there too and will raise once the job moves to a newer JDK.
-
-Operations on a set you already hold are unaffected, since they carry the key the set was
-obtained under. That is why a memory set may be handed to a worker thread but the long-term
-memory itself may not.
+Call `getMemorySet` / `get_memory_set` and `deleteMemorySet` / `delete_memory_set` in
+the Action body, outside functions submitted for async or durable execution. Operations
+such as `add`, `get`, `search`, and `delete` on a `MemorySet` obtained in that Action can
+run in those functions. See
+[Using Resources and Memory]({{< ref "docs/development/create-agent/actions#using-resources-and-memory" >}})
+for the access rules and an async search example.
Long-term memory also requires a non-empty partition key. A key of `""` carries no
isolation in the backing store, so obtaining or using a memory set under one raises
diff --git a/docs/content/docs/development/memory/sensory_and_short_term_memory.md b/docs/content/docs/development/memory/sensory_and_short_term_memory.md
index e15df2c1..a3e3e9ed 100644
--- a/docs/content/docs/development/memory/sensory_and_short_term_memory.md
+++ b/docs/content/docs/development/memory/sensory_and_short_term_memory.md
@@ -35,6 +35,11 @@ Short-Term Memory is shared across all actions within an agent run, and multiple
## When to Use
+Access Sensory and Short-Term Memory in the Action body, before or after awaiting a call.
+Do not access them from a function submitted for async or durable execution, including
+through a previously obtained `MemoryObject`. See
+[Using Resources and Memory]({{< ref "docs/development/create-agent/actions#using-resources-and-memory" >}}).
+
### Sensory Memory
Sensory Memory is ideal for:
diff --git a/docs/content/docs/get-started/overview.md b/docs/content/docs/get-started/overview.md
index c8bbfee0..4b52aa5c 100644
--- a/docs/content/docs/get-started/overview.md
+++ b/docs/content/docs/get-started/overview.md
@@ -41,6 +41,13 @@ The key features of Apache Flink Agents include:
## Getting Started
+Agents use Actions to process Events and Resources to provide reusable, named runtime
+components such as models, tools, prompts, and MCP servers. See
+[Core Concepts]({{< ref "docs/development/create-agent/overview#core-concepts" >}})
+for how these pieces work together, and
+[Registering and Referencing Resources]({{< ref "docs/development/create-agent/overview#registering-and-referencing-resources" >}})
+for how Actions access Resources.
+
Start with [Setup]({{< ref "docs/get-started/setup" >}}), then choose a Quickstart:
- [Build an Agent]({{< ref "docs/get-started/quickstart/create-agent" >}})
diff --git a/docs/content/docs/get-started/quickstart/create-agent/_index.md b/docs/content/docs/get-started/quickstart/create-agent/_index.md
index c2d61c52..132b2658 100644
--- a/docs/content/docs/get-started/quickstart/create-agent/_index.md
+++ b/docs/content/docs/get-started/quickstart/create-agent/_index.md
@@ -27,9 +27,7 @@ under the License.
## Overview
-An Agent in Flink-Agents is an agent whose reasoning and behavior are organized as a directed workflow of modular steps, called actions, connected by events. This design is inspired by the need to orchestrate complex, multi-stage tasks in a transparent, extensible, and data-centric way, leveraging Apache Flink's streaming architecture.
-
-This quickstart introduces two small, progressive streaming examples that demonstrate how to build LLM-powered workflows with Flink Agents:
+This quickstart walks through two examples of building AI agents that process streaming data with Flink Agents:
- **Review Analysis**: Processes a stream of product reviews and uses a single agent to extract a rating (1–5) and unsatisfied reasons from each review.
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 555fac8d..1fa99c34 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
@@ -28,6 +28,10 @@ This example builds on [Build Your First Agent]({{< ref "docs/get-started/quicks
Reuse the [environment and connection setup]({{< ref "docs/get-started/quickstart/create-agent/single-agent#prepare-agents-execution-environment" >}}) and [example preparation]({{< ref "docs/get-started/quickstart/create-agent/single-agent#run-the-example" >}}) from the first walkthrough.
+Both Agents can reference the connection Resource registered on that environment by name.
+See [Registering and Referencing Resources]({{< ref "docs/development/create-agent/overview#registering-and-referencing-resources" >}})
+for how environment-level Resources are made available to Agents.
+
The code for the `ProductSuggestionAgent`, which is used to generate product improvement suggestions based on the aggregated analysis results, is similar to the `ReviewAnalysisAgent`.
### Read Input via the Table API
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 526b7108..30b75c42 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
@@ -30,7 +30,11 @@ Build an Agent that analyzes a stream of product reviews.
### Prepare Agents Execution Environment
-Create the agents execution environment, and register the available chat model connections, which can be used by the agents, to the environment.
+Create the agents execution environment and register an Ollama connection as a named
+Resource of type `CHAT_MODEL_CONNECTION`. Registering it on the environment makes it available
+to the Agents in that environment. See
+[Registering and Referencing Resources]({{< ref "docs/development/create-agent/overview#registering-and-referencing-resources" >}})
+for the shared registration and access model.
{{< tabs "Prepare Agents Execution Environment" >}}
@@ -69,7 +73,14 @@ agentsEnv.addResource(
### Create the Agents
-Below is the example code for the `ReviewAnalysisAgent`, which is used to analyze the product reviews and generate a satisfaction score and potential reasons for dissatisfaction. It demonstrates how to define the prompt, tool, chat model, and action for the agent. Also, it shows how to process the chat response and send the output event. For more details, please refer to the [Agent]({{< ref "docs/development/create-agent/overview" >}}) documentation.
+`ReviewAnalysisAgent` analyzes product reviews and produces a satisfaction score and reasons
+for dissatisfaction. It declares prompt, tool, and chat model Resources, along with two Actions.
+The chat model setup references the environment's connection by name: `ollama_server` in Python
+or `ollamaChatModelConnection` in Java.
+
+The input Action sends a `ChatRequestEvent` naming the model Resource (`review_analysis_model`
+or `reviewAnalysisModel`). The [built-in chat Action]({{< ref "docs/development/calling-llms/chat#chat-events" >}})
+gets and calls that Resource, then sends a response Event for the output Action to handle.
{{< tabs "Create the Agents" >}}
diff --git a/docs/content/docs/get-started/quickstart/parallel_llm.md b/docs/content/docs/get-started/quickstart/parallel_llm.md
index d140bb6b..7f820348 100644
--- a/docs/content/docs/get-started/quickstart/parallel_llm.md
+++ b/docs/content/docs/get-started/quickstart/parallel_llm.md
@@ -35,7 +35,11 @@ The **Parallel Sentiment Analysis** agent processes a restaurant review and judg
### Prepare Agents Execution Environment
-Create the agents execution environment, and register the available chat model connection to the environment.
+Create the agents execution environment and register the Ollama connection as a named
+Resource of type `CHAT_MODEL_CONNECTION`. The Agent's model setup references this connection
+as `ollama_server` in Python or `ollamaChatModelConnection` in Java. See
+[Registering and Referencing Resources]({{< ref "docs/development/create-agent/overview#registering-and-referencing-resources" >}})
+for the shared registration and access model.
{{< tabs "Prepare Agents Execution Environment" >}}
@@ -81,6 +85,11 @@ agentsEnv.addResource(
### Create the Agent
+The model Resource is named `sentiment_model` in Python or `sentimentModel` in Java.
+Each `ChatRequestEvent` references that name; the
+[built-in chat Action]({{< ref "docs/development/calling-llms/chat#chat-events" >}})
+gets the Resource and performs the model call.
+
Below is the example code for the `ParallelChatAgent`. Two system prompts — `PARALLEL_SYSTEM_PROMPT` for the parallel aspect judgments and `AGGREGATE_SYSTEM_PROMPT` for the aggregation call — are packaged into `ChatRequestEvent`s by `_build_aspect_request` / `buildAspectRequest` and `_build_summarize_request` / `buildSummarizeRequest`. The agent defines a chat model and four actions: `request_aspect_judgments` stores the review `id` and `text` in sensory memory and emits a single generic [...]
{{< tabs "Create the Agent" >}}
@@ -167,7 +176,7 @@ class ParallelChatAgent(Agent):
def request_aspect_judgments(event: Event, ctx: RunnerContext) -> None:
"""Process input event and dispatch SentimentInputEvent to aspect handlers."""
payload = InputEvent.from_event(event).input
- # Primitive types (int, str) cross the Pemja JVM boundary without serialization.
+ # Store the input's primitive values in Sensory Memory.
ctx.sensory_memory.set("id", payload["id"])
ctx.sensory_memory.set("text", payload["text"])
ctx.send_event(
@@ -314,9 +323,9 @@ Other private helpers (`buildOutputEvent`) are omitted above; the full source is
{{< hint warning >}}
**Constraints when using parallel actions:**
-- **Dispatch action must exit immediately after sending events.** The action that emits `SentimentInputEvent` (i.e. `request_aspect_judgments`) should return right after calling `ctx.send_event(...)`. Do not `await`, block, or perform further business logic in the same action — the framework needs control back to schedule the parallel chat actions.
-- **Action handlers on the same key execute sequentially.** Multiple action invocations triggered by the same key are processed one at a time, not concurrently. This means `handle_taste_input` and `handle_service_input` are called in sequence. Each handler writes a single path-based entry (`aspect_map.<request_id>`) to sensory memory without reading back, so there is no conflict between the two. The actual LLM calls dispatched by these handlers still execute in parallel.
-- **Response handlers on the same key execute sequentially.** Multiple `ChatResponseEvent` triggers for the same key are processed one at a time, not concurrently. This means the path-based writes to `sentiments.<aspect>` in `handle_response` are safe and will not suffer from concurrent overwrites.
+- **Finish dispatch before waiting for downstream results.** The `request_aspect_judgments` Action's `SENTIMENT_INPUT_EVENT_TYPE` Event is routed after that Action finishes. Awaiting unrelated work can let other ready Actions run, but does not dispatch this buffered Event. Keep the dispatch Action short.
+- **Keep shared-memory updates in the Action body.** The request and response handlers in this example update Sensory Memory without awaiting between reading and writing it. If you add an await, another Action may update shared Memory before the handler resumes. Do not move Memory access into the asynchronous call itself.
+- **Check the runtime requirements for concurrent calls.** See [Async Execution]({{< ref "docs/development/create-agent/actions#async-execution" >}}) and [Execution Requirements]({{< ref "docs/development/create-agent/actions#execution-requirements" >}}) for scheduling behavior and setup.
{{< /hint >}}
### Integrate the Agent with Flink
diff --git a/docs/content/docs/get-started/quickstart/react_agent.md b/docs/content/docs/get-started/quickstart/react_agent.md
index ad678533..27c28c5a 100644
--- a/docs/content/docs/get-started/quickstart/react_agent.md
+++ b/docs/content/docs/get-started/quickstart/react_agent.md
@@ -36,7 +36,11 @@ The **Review Analysis** agent processes a stream of product reviews and uses a s
### Prepare Agents Execution Environment
-Create the agents execution environment, and register the available chat model connections and tools to the environment.
+Create the agents execution environment and register the Ollama connection and notification
+tool as named Resources of types `CHAT_MODEL_CONNECTION` and `TOOL`. The ReAct Agent's model
+configuration references these names through its `connection` and `tools` fields. See
+[Registering and Referencing Resources]({{< ref "docs/development/create-agent/overview#registering-and-referencing-resources" >}})
+for the shared registration and access model.
{{< tabs "Prepare Agents Execution Environment" >}}
diff --git a/docs/content/docs/get-started/quickstart/yaml_agent.md b/docs/content/docs/get-started/quickstart/yaml_agent.md
index d0253cac..68607061 100644
--- a/docs/content/docs/get-started/quickstart/yaml_agent.md
+++ b/docs/content/docs/get-started/quickstart/yaml_agent.md
@@ -24,7 +24,11 @@ under the License.
## Overview
-The YAML API lets you declare a Flink Agents application — the agent's actions, tools, prompts, chat models, and other resources — as a single declarative file, and load it into an `AgentsExecutionEnvironment` with one call. Your action and tool implementations still live in Python or Java; the YAML file only describes how those pieces are wired together.
+The YAML API lets you declare an Agent's Actions and named Resources in a file and load it
+into an `AgentsExecutionEnvironment`. It uses the same
+[core concepts]({{< ref "docs/development/create-agent/overview#core-concepts" >}})
+as the Java and Python APIs. Your Action and tool implementations still live in Python or Java;
+the YAML file declares their configuration and connections.
This quickstart runs the same **Review Analysis** workflow as the [Build Your First Agent]({{< ref "docs/get-started/quickstart/create-agent/single-agent" >}}) — extracting a satisfaction score and dissatisfaction reasons from a stream of product reviews — but declares the agent in YAML and loads it with `load_yaml` / `loadYaml`. The action functions and the `notify_shipping_manager` tool are reused from the single-Agent example as static methods, so this quickstart shows the **smallest [...]
@@ -187,10 +191,17 @@ agents:
A few things to notice in the YAML above:
+- A Resource's `name` is the name used by Actions and other Resources to find it. Its section
+ determines the resource type; for example, `chat_model_setups` declares chat model Resources.
+ See [Registering and Referencing Resources]({{< ref "docs/development/create-agent/overview#registering-and-referencing-resources" >}}).
+- The chat model setup's `connection`, `prompt`, and `tools` fields reference registered
+ Resource names. Its `model: qwen3:8b` field selects the model on the Ollama server.
+- The reused input Action sends a `ChatRequestEvent` naming the chat model setup. The
+ [built-in chat Action]({{< ref "docs/development/calling-llms/chat#chat-events" >}})
+ gets and calls that Resource, then sends a response Event.
- `clazz: ollama` is an alias resolved by the loader to the full Ollama chat-model class — see the alias table in the [YAML API]({{< ref "docs/development/create-agent/yaml#class-aliases" >}}) doc.
- `trigger_conditions: [input]` / `[chat_response]` use **event aliases** for the framework's built-in events. An alias replaces only a complete event-type entry; it never rewrites text inside a condition expression or quoted event type.
- `function:` strings use the `<module-or-class>:<qualname>` format. The right side is the class-qualified method name, so the YAML reuses the same `process_input` / `processInput` static methods the original `ReviewAnalysisAgent` already defines.
-- The prompt is declared inline as a `messages:` list and referenced from the chat-model setup by name.
### Load and Run
diff --git a/docs/content/docs/get-started/setup.md b/docs/content/docs/get-started/setup.md
index f15fc120..8095fe11 100644
--- a/docs/content/docs/get-started/setup.md
+++ b/docs/content/docs/get-started/setup.md
@@ -52,7 +52,7 @@ Flink release you deploy. Concurrency and integration requirements depend on the
feature:
- Java Action scheduling differs between JDK 21+ and earlier JDKs. See
- [Async Execution]({{< ref "docs/development/create-agent/actions#async-execution" >}})
+ [JDK and Scheduling Requirements]({{< ref "docs/development/create-agent/actions#jdk-and-scheduling-requirements" >}})
for configuration and the JDK 21+ JVM option.
- Java-native MCP requires JDK 17+; on earlier supported JDKs, it uses the Python SDK
and requires its Python dependencies. See [MCP SDK]({{< ref "docs/development/calling-llms/mcp#mcp-sdk" >}}).
diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md
index a11782ab..d9bc526b 100644
--- a/docs/content/docs/operations/configuration.md
+++ b/docs/content/docs/operations/configuration.md
@@ -126,7 +126,7 @@ Here is the list of all built-in core configuration options.
| Key | Default | Type | Description |
|---------------------------|----------------------------|-----------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `event-listeners` | none | `List<String>` | The list of event listener class names. Each class must implement the EventListener interface and provide a public no-argument constructor. {{< hint warning >}} Note: Currently, custom event listeners are only supported in Java. {{< /hint >}} |
-| `action.trigger-condition.evaluate-failure-strategy` | `WARN_AND_SKIP` | ConditionEvaluationFailureStrategy | Handles event-time failures while preparing variables for or evaluating a compiled condition, including a dynamic non-Boolean result. <br/><ul><li>`WARN_AND_SKIP` (default): log a warning, treat that condition as false, and continue with later OR conditions.</li><li>`FAIL`: throw `IllegalStateException` and fail the Flink task; recovery follows the job's restart configuration.< [...]
+| `action.trigger-condition.evaluate-failure-strategy` | `WARN_AND_SKIP` | ConditionEvaluationFailureStrategy | Handles failures while evaluating a trigger condition for an individual Event, including missing attributes, incompatible values, or a non-Boolean result. <br/><ul><li>`WARN_AND_SKIP` (default): log a warning, treat that condition as false, and continue with later OR conditions.</li><li>`FAIL`: throw `IllegalStateException` and fail the Flink task; recovery follows the job's re [...]
| `max-retries` | 0 | int | Number of additional attempts per model call, including routing judge calls. Defaults to 0 (no retries). |
| `retry-wait-interval` | 1 | int | Base wait interval in seconds between retries. Uses exponential backoff: the actual wait time for the Nth retry is `retry-wait-interval * 2^(N-1)` seconds. For example, with default 1s, waits are 1s, 2s, 4s, etc. Retry count and total wait time are reported in `ChatResponseEvent` and recorded as metrics (`retryCount`, `retryWaitSec`) under the configured ChatModel resource name. |
| `chat.async` | true | boolean | Whether chat asynchronously for built-in chat action. |
diff --git a/docs/content/docs/operations/deployment.md b/docs/content/docs/operations/deployment.md
index 7c5dd7c3..8d9ee4e9 100644
--- a/docs/content/docs/operations/deployment.md
+++ b/docs/content/docs/operations/deployment.md
@@ -97,11 +97,20 @@ After recovery from a checkpoint, Flink Agents reprocess events that arrived aft
### Exactly-Once Action Consistency
-To ensure exactly-once action consistency, you must configure an external action state store. Flink Agents record action state in this store on a per-action basis. After recovering from a checkpoint, Flink Agents consult the external store and will not re-execute actions that were already completed. This guarantees each action is executed exactly once after recovering from a checkpoint.
+To enable exactly-once action consistency, configure an external action state store.
+Flink Agents record state for each Action in this store. After recovering from a
+checkpoint, the framework reuses Actions whose completion was persisted, including
+their stored Sensory/Short-Term Memory updates and output Events. Unfinished Actions
+may execute again.
When a completed Action is reused during recovery, its stored output Events keep their original Event IDs. The runtime records their relationships to the current trigger Event in Trace Log `entityMetadata`.
-The same persisted action state is also used by fine-grained durable execution.
+The same store also saves individual call outcomes for
+[Durable Execution]({{< ref "docs/development/create-agent/actions#durable-execution" >}}).
+When an unfinished Action runs again, durable calls can reuse persisted outcomes.
+An external operation can finish before its outcome is saved, so recovery may repeat
+that operation. Use idempotent operations or reconciliation for external side effects;
+see [Recovery Behavior]({{< ref "docs/development/create-agent/actions#recovery-behavior" >}}).
{{< hint info >}}
**Note**: Currently, Kafka and Fluss are supported as the external action state store.