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 6a6a5b33 [docs] Document Action trigger conditions (#977)
6a6a5b33 is described below
commit 6a6a5b336ae46ad042a047ca3c2d1104620f5cd1
Author: Kerui Wang <[email protected]>
AuthorDate: Tue Aug 18 20:47:08 2026 +0800
[docs] Document Action trigger conditions (#977)
Generated-by: Codex CLI 0.144.5 (GPT-5.6-sol)
Co-authored-by: WenjinXie <[email protected]>
---
docs/content/docs/development/workflow_agent.md | 179 ++++++++++++++++++---
docs/content/docs/development/yaml.md | 33 +++-
.../docs/get-started/quickstart/yaml_agent.md | 2 +-
docs/content/docs/operations/configuration.md | 1 +
4 files changed, 186 insertions(+), 29 deletions(-)
diff --git a/docs/content/docs/development/workflow_agent.md
b/docs/content/docs/development/workflow_agent.md
index 10b1c226..b8166cd0 100644
--- a/docs/content/docs/development/workflow_agent.md
+++ b/docs/content/docs/development/workflow_agent.md
@@ -26,7 +26,12 @@ under the License.
A workflow style 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.
-In Flink-Agents, a workflow agent is defined as a class that inherits from the
`Agent` base class. The agent's logic is expressed as a set of actions, each of
which is a function decorated with `@action(EventType.X)` in python (or a
method annotated with `@Action(EventType.X)` in java). Actions consume events,
perform reasoning or tool calls, and emit new events, which may trigger
downstream actions. This event-driven workflow forms a directed cyclic graph of
computation, where each node [...]
+In Flink-Agents, a workflow 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.
A workflow 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.
@@ -230,12 +235,14 @@ public class ReviewAnalysisAgent extends Agent {
## Action
-An action is a piece of code that can be executed. Each action listens to at
least one type of event. When an event of the listening type occurs, the action
will be triggered. An action can also generate new events, to trigger other
actions.
+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.
-
-To declare an action in Agent, user can use `@action` to decorate a function
of Agent class in python (or annotate a method of Agent class in java), and
declare the listened event types as decorator/annotation parameters.
-
-The decorated/annotated function signature should be `(Event, RunnerContext)
-> None`. In Python, actions can also be defined as `async def` when using
async execution (see [Async Execution](#async-execution)).
+Use `@action(*trigger_conditions, target=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)).
{{< tabs "Action Function" >}}
@@ -264,9 +271,78 @@ public class ReviewAnalysisAgent extends Agent {
{{< /tabs >}}
-In the function, user can also send new events, to trigger other actions, or
output the data.
+### Registering Actions Programmatically
+
+As an alternative to annotations, register an action through the `Agent` API:
+
+{{< tabs "Programmatic Action" >}}
+
+{{< tab "Python" >}}
+```python
+agent.add_action(
+ "process_event",
+ [EventType.InputEvent],
+ process_event,
+)
+```
+{{< /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));
+ }
+
+ public static void processEvent(Event event, RunnerContext ctx) {
+ // Handle the input event.
+ }
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+### Trigger Conditions
+
+A trigger condition is either:
+
+- 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).
+
+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:
+
+```text
+type == EventType.InputEvent && attributes.urgent == true
+```
-**Trigger another action** — send a built-in or custom event that another
action listens to:
+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`.
+
+#### Matching Behavior
+
+When an event arrives:
+
+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.
+
+### Sending Events
+
+An action can send an event to trigger another action or emit output
downstream.
+
+**Trigger another action** — send a built-in or custom event that matches
another action's trigger
+conditions:
{{< tabs "Trigger Another Action" >}}
@@ -275,7 +351,6 @@ In the function, user can also send new events, to trigger
other actions, or out
@action(EventType.InputEvent)
@staticmethod
def process_input(event: Event, ctx: RunnerContext) -> None:
- # send a ChatRequestEvent to trigger the built-in chat-model action
ctx.send_event(ChatRequestEvent(model="my_model", messages=messages))
```
{{< /tab >}}
@@ -284,8 +359,6 @@ def process_input(event: Event, ctx: RunnerContext) -> None:
```java
@Action(EventType.InputEvent)
public static void processInput(Event event, RunnerContext ctx) throws
Exception {
- InputEvent inputEvent = InputEvent.fromEvent(event);
- // send ChatRequestEvent
ctx.sendEvent(new ChatRequestEvent("my_model", messages));
}
```
@@ -293,16 +366,15 @@ public static void processInput(Event event,
RunnerContext ctx) throws Exception
{{< /tabs >}}
-**Emit downstream output** — send an `OutputEvent` to produce an output of the
agent:
+**Emit downstream output** — send an `OutputEvent`:
{{< tabs "Emit Output" >}}
{{< tab "Python" >}}
```python
-@action(ChatResponseEvent.EVENT_TYPE)
+@action(EventType.ChatResponseEvent)
@staticmethod
def emit_output(event: Event, ctx: RunnerContext) -> None:
- # output data to downstream
ctx.send_event(OutputEvent(output=result))
```
{{< /tab >}}
@@ -311,7 +383,6 @@ def emit_output(event: Event, ctx: RunnerContext) -> None:
```java
@Action(EventType.ChatResponseEvent)
public static void emitOutput(Event event, RunnerContext ctx) {
- // output data to downstream
ctx.sendEvent(new OutputEvent(result));
}
```
@@ -320,14 +391,69 @@ public static void emitOutput(Event event, RunnerContext
ctx) {
{{< /tabs >}}
{{< hint info >}}
-An `OutputEvent` is collected and emitted to the agent's downstream
**immediately**, bypassing
-action routing, while other events (such as `ChatRequestEvent`) are routed to
the actions that
-listen for them. Sending a `ChatRequestEvent` and an `OutputEvent` from the
same action is valid
-API usage, but it produces both an immediate output and, once the chat
response is handled, a
-later model-based output. For the normal chat request/response workflow, emit
the `OutputEvent`
-from the action that handles the `ChatResponseEvent`, as shown above.
+`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
+
+#### 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. 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.
+
+#### Expression Variables
+
+Java and Python actions use the same Java runtime to evaluate condition
expressions. Expressions can
+access these framework variables:
+
+| 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 |
+
+#### Accessing Attributes
+
+- 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.content`.
+- 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`.
+
+#### Missing Attributes and Value Types
+
+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" >}}).
+
+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.
+
+#### Limitations
+
+- `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.
+
### Durable Execution
Use durable execution when you wrap a time-consuming or side-effecting
operation. The framework persists the result and replays it on recovery when
the same call is encountered, so the function will not be called again and side
effects are avoided. 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.
@@ -580,12 +706,17 @@ public class MyAgent extends Agent {
## Event
-Events are JSON-serializable messages passed between actions. Every event has
a `type` string used for routing and an `attributes` map that carries the
payload. A single event may trigger multiple actions if they are all listening
to its type.
+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.
### Special Events
-* `InputEvent`: Generated by the framework, carrying an input data record that
arrives at the agent in its `input` attribute. Actions listening to
`InputEvent` are the entry points of the agent.
-* `OutputEvent`: The framework listens to `OutputEvent` and converts its
`output` attribute into outputs of the agent.
+* `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.
### Unified Event
diff --git a/docs/content/docs/development/yaml.md
b/docs/content/docs/development/yaml.md
index b70653f7..b4837566 100644
--- a/docs/content/docs/development/yaml.md
+++ b/docs/content/docs/development/yaml.md
@@ -278,19 +278,38 @@ Inline action (map) fields:
|-------|----------|-------------|
| `name` | yes | Action name (unique within the agent). |
| `function` | yes | Fully-qualified callable in the form
`<module-or-class>:<qualname>`. See [Function
references](#function-references). |
-| `trigger_conditions` | yes | List of event types the action listens to.
Built-in [event aliases](#event-aliases) (`input`, `chat_request`, ...) or your
own event-type strings. |
+| `trigger_conditions` | yes | Non-empty list of strings containing exact
event types or condition expressions that return a Boolean. Each string must
contain at least one non-whitespace character. Built-in [event
aliases](#event-aliases) (`input`, `chat_request`, ...) may be used as complete
entries. All entries use OR semantics. |
| `type` | no | Implementation language: `python` or `java`. Defaults to
`python` (see [Selecting the implementation
language](#selecting-the-implementation-language)). |
| `config` | no | Free-form configuration map passed to the action at runtime.
|
+For built-in event types, prefer an [event alias](#event-aliases) such as
`input`. Trigger-condition
+matching and expression semantics are shared with annotated and
programmatically registered actions;
+see [Trigger Conditions]({{< ref
"docs/development/workflow_agent#trigger-conditions" >}}).
+
+Every entry must remain a non-empty string after YAML parsing. Quote condition
expressions, for
+example `"ready == true"`. Quotes are required for literal expressions such as
`"true"`; without
+them, YAML parses `true`, `false`, or `null` as a non-string value. When an
exact custom event type
+requires quotes, preserve those quotes inside the YAML string:
+
+```yaml
+trigger_conditions:
+ - order-created
+ - "'order:created'"
+ - "'EventType.custom'"
+```
+
+An `actions` list can mix inline actions and shared-action references:
+
```yaml
actions:
- name: action1
function: my_pkg.actions:action1
- trigger_conditions: [input]
+ trigger_conditions: [input, custom_event]
type: python
- name: action2
function: my_pkg.actions:action2
- trigger_conditions: [chat_response]
+ trigger_conditions:
+ - "type == EventType.ChatResponseEvent && response.content != ''"
type: python
- action3 # shared action reference (declared at file
level)
```
@@ -498,7 +517,7 @@ When `type:` resolves to the **opposite** language of the
loader, the loader bui
### Provider aliases
-For `clazz:` on resource descriptors and for event names in
`trigger_conditions:`, you can use a short alias instead of a fully-qualified
class path.
+For `clazz:` on resource descriptors and for complete event-type entries in
`trigger_conditions:`, you can use a short alias instead of a fully-qualified
class path. Event alias replacement is an exact complete-entry lookup: `input`
is replaced, while `type == input`, `attributes.kind == 'input'`, and the
quoted event type `'input'` remain unchanged.
#### Event aliases
@@ -513,6 +532,12 @@ For `clazz:` on resource descriptors and for event names
in `trigger_conditions:
| `context_retrieval_request` | `ContextRetrievalRequestEvent` |
| `context_retrieval_response` | `ContextRetrievalResponseEvent` |
+{{< hint warning >}}
+`output` resolves to `OutputEvent`, but output events are emitted directly
downstream and bypass
+action matching. Therefore, using `output` as a trigger condition never
invokes an action. See
+[Sending events]({{< ref "docs/development/workflow_agent#sending-events" >}}).
+{{< /hint >}}
+
For custom event types defined in your code, write the event's full
`EVENT_TYPE` string instead of an alias.
#### Class aliases
diff --git a/docs/content/docs/get-started/quickstart/yaml_agent.md
b/docs/content/docs/get-started/quickstart/yaml_agent.md
index 82fb55bc..949995c4 100644
--- a/docs/content/docs/get-started/quickstart/yaml_agent.md
+++ b/docs/content/docs/get-started/quickstart/yaml_agent.md
@@ -188,7 +188,7 @@ agents:
A few things to notice in the YAML above:
- `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/yaml#class-aliases" >}}) doc.
-- `trigger_conditions: [input]` / `[chat_response]` use **event aliases** for
the framework's built-in events.
+- `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.
diff --git a/docs/content/docs/operations/configuration.md
b/docs/content/docs/operations/configuration.md
index 2dc17981..c1918bb5 100644
--- a/docs/content/docs/operations/configuration.md
+++ b/docs/content/docs/operations/configuration.md
@@ -127,6 +127,7 @@ Here is the list of all built-in core configuration options.
| `baseLogDir` | (none) | String
| Base directory for file-based event logs. If not set, uses
`java.io.tmpdir/flink-agents`. Setting this value also implicitly switches
`eventLoggerType` to `file`.
|
| `prettyPrint` | false | boolean
| Whether to enable pretty-printed JSON format for event logs. When set to
`true`, each event is written as formatted multi-line JSON instead of JSONL
(JSON Lines) format. {{< hint info >}}Note: enabling this option makes the log
file no longer valid JSONL format. {{< /hint >}} |
| `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.< [...]
| `error-handling-strategy` | ErrorHandlingStrategy.FAIL |
ErrorHandlingStrategy | Strategy for handling errors during model requests,
include timeout and unexpected output schema. <br/>The option value could
be:<br/> <ul><li>`ErrorHandlingStrategy.FAIL`</li>
<li>`ErrorHandlingStrategy.RETRY`</li> <li>`ErrorHandlingStrategy.IGNORE`</li> |
| `max-retries` | 3 | int
| Number of retries when using `ErrorHandlingStrategy.RETRY`.
|
| `retry-wait-interval` | 1 | int
| Base wait interval in seconds between retries when using
`ErrorHandlingStrategy.RETRY`. 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 connection name. |
