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 <[email protected]>
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 <[email protected]>
---
.../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.
