Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions api/src/main/java/org/apache/flink/agents/api/EventType.java
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@

package org.apache.flink.agents.api;

import java.util.Map;

/**
* Compile-time constants for built-in event types, sourced from each {@code XxxEvent.EVENT_TYPE}.
*
Expand All @@ -40,5 +42,23 @@ public final class EventType {
public static final String ContextRetrievalResponseEvent =
org.apache.flink.agents.api.event.ContextRetrievalResponseEvent.EVENT_TYPE;

private static final Map<String, String> ALL_CONSTANTS =
Map.of(
"InputEvent", InputEvent,
"OutputEvent", OutputEvent,
"ChatRequestEvent", ChatRequestEvent,
"ChatResponseEvent", ChatResponseEvent,
"ToolRequestEvent", ToolRequestEvent,
"ToolResponseEvent", ToolResponseEvent,
"ContextRetrievalRequestEvent", ContextRetrievalRequestEvent,
"ContextRetrievalResponseEvent", ContextRetrievalResponseEvent);

/**
* Returns the built-in event type constants as a name-to-value map for condition expressions.
*/
public static Map<String, String> allConstants() {
return ALL_CONSTANTS;
}

private EventType() {}
}
15 changes: 8 additions & 7 deletions api/src/main/java/org/apache/flink/agents/api/agents/Agent.java
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;

/** Base class for defining agent logic. */
Expand All @@ -42,7 +43,7 @@ public Agent() {
for (ResourceType type : ResourceType.values()) {
this.resources.put(type, new HashMap<>());
}
this.actions = new HashMap<>();
this.actions = new LinkedHashMap<>();
}

public Map<String, Tuple3<String[], Function, Map<String, Object>>> getActions() {
Expand All @@ -56,8 +57,8 @@ public Map<ResourceType, Map<String, Object>> getResources() {
/**
* Add action to agent.
*
* @param triggerConditions Trigger condition strings — each is either an event-type name or a
* future condition-expression form.
* @param triggerConditions Raw event-type names or Boolean conditions combined with OR
* semantics. Shape and expression validation occur during {@code AgentPlan} construction.
* @param method The method of this action, should be static method.
* @param config The optional config can be used by this action.
*/
Expand All @@ -70,8 +71,8 @@ public Agent addAction(
/**
* Add action to agent.
*
* @param triggerConditions Trigger condition strings — each is either an event-type name or a
* future condition-expression form.
* @param triggerConditions Raw event-type names or Boolean conditions combined with OR
* semantics. Shape and expression validation occur during {@code AgentPlan} construction.
* @param method The method of this action, should be static method.
*/
public Agent addAction(String[] triggerConditions, Method method) {
Expand All @@ -82,8 +83,8 @@ public Agent addAction(String[] triggerConditions, Method method) {
* Add action to agent.
*
* @param name The action name. Must be unique within this agent.
* @param triggerConditions Trigger condition strings — each is either an event-type name or a
* future condition-expression form.
* @param triggerConditions Raw event-type names or Boolean conditions combined with OR
* semantics. Shape and expression validation occur during {@code AgentPlan} construction.
* @param function The api-layer function descriptor; will be promoted to a plan-layer
* executable at {@code AgentPlan} construction.
* @param config Optional config for this action.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,46 +24,31 @@
import java.lang.annotation.Target;

/**
* Marks a method as an agent action triggered by matching events.
* Marks a Java method as an agent action triggered by event types or condition expressions.
*
* <p>Each {@link #value()} entry is an event type name string. Use the {@code EVENT_TYPE} constants
* on built-in event classes, the {@link org.apache.flink.agents.api.EventType} constants, or plain
* strings for custom events. Multiple entries combine with OR.
* <p>Each {@link #value()} entry is either an exact event-type name or a condition expression that
* evaluates to boolean. Multiple entries use OR semantics. An entry matching the event-type syntax
* is classified as an event type rather than as a condition expression. To combine an event-type
* restriction and an attribute predicate using AND, place both in a single expression, for example
* {@code type == EventType.InputEvent && score > 5}.
*
* <pre>{@code
* // Built-in event type via the EventType constant
* @Action(EventType.InputEvent)
* <p>The event-type syntax is a bare or dotted name, such as {@code order.created} or {@code
* order-created}; each segment may contain hyphens. Boolean attribute conditions must therefore be
* explicit, for example {@code ready == true}.
*
* // Equivalent via the legacy class constant
* @Action(InputEvent.EVENT_TYPE)
* <p>The API preserves entries as raw strings. Entries are classified and condition expressions are
* validated when the agent plan is built.
*
* // User-defined event type
* @Action("MyCustomEvent")
*
* // Multiple types (OR semantics)
* @Action({EventType.InputEvent, "MyCustomEvent"})
* }</pre>
*
* <p>For a cross-language action, set {@link #target()} to a {@link PythonFunction} with a
* non-empty {@code module}. The annotated Java body is never invoked — throw {@link
* UnsupportedOperationException} so direct calls outside the framework fail loud:
*
* <pre>{@code
* @Action(
* value = EventType.InputEvent,
* target = @PythonFunction(module = "my_pkg.handlers", qualname = "handle_input"))
* public void handleInput(Event event, RunnerContext ctx) {
* throw new UnsupportedOperationException("cross-language stub");
* }
* }</pre>
* <p>Set {@link #target()} to a {@link PythonFunction} configured with a module for a
* cross-language action. The annotated Java method is not invoked for such actions.
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Action {
/**
* Event type name strings; multiple entries have OR semantics. Named {@code value} (not {@code
* triggerConditions}) to enable the {@code @Action({...})} shorthand (JLS §9.7.3); corresponds
* to Python's {@code *trigger_conditions}.
* Raw event-type names or explicit Boolean conditions combined with OR semantics. Named {@code
* value} to enable the {@code @Action({...})} shorthand (JLS §9.7.3); corresponds to Python's
* {@code *trigger_conditions}.
*/
String[] value();

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,12 @@
/** The set of configuration options for agents parameters. */
public class AgentConfigOptions {

/** Behaviour when a trigger condition throws or returns a non-Boolean at evaluation time. */
public enum ConditionEvaluationFailureStrategy {
WARN_AND_SKIP,
FAIL
}

/**
* The config parameter specifies which event logger implementation to use. Defaults to {@link
* LoggerType#SLF4J}, which surfaces events in Flink's Web UI; setting {@link LoggerType#FILE}
Expand All @@ -33,6 +39,17 @@ public class AgentConfigOptions {
public static final ConfigOption<LoggerType> EVENT_LOGGER_TYPE =
new ConfigOption<>("eventLoggerType", LoggerType.class, LoggerType.SLF4J);

/**
* Specifies how condition evaluation failures are handled. Defaults to {@code WARN_AND_SKIP};
* use {@code FAIL} to enforce fail-fast behavior.
*/
public static final ConfigOption<ConditionEvaluationFailureStrategy>
CONDITION_EVALUATION_FAILURE_STRATEGY =
new ConfigOption<>(
"action.trigger-condition.evaluate-failure-strategy",
ConditionEvaluationFailureStrategy.class,
ConditionEvaluationFailureStrategy.WARN_AND_SKIP);

/** The config parameter specifies the directory for the FileEvent file. */
public static final ConfigOption<String> BASE_LOG_DIR =
new ConfigOption<>("baseLogDir", String.class, null);
Expand Down
43 changes: 25 additions & 18 deletions api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java
Original file line number Diff line number Diff line change
Expand Up @@ -321,27 +321,31 @@ public static void loadYaml(AgentsExecutionEnvironment env, List<Path> paths) {
for (Path path : paths) {
LoadedFile loaded = buildAgents(path);

// Resolve shared-action string refs first so a bad ref doesn't leave partial state on
// the env.
for (Map.Entry<String, Agent> entry : loaded.getAgents().entrySet()) {
AgentSpec spec = loaded.getAgentSpecs().get(entry.getKey());
List<ActionSpec> orderedActions = new ArrayList<>();
for (AgentActionRef ref : spec.getActions()) {
if (!ref.isReference()) {
continue;
ActionSpec actionSpec = ref.getSpec();
if (ref.isReference()) {
actionSpec = loaded.getSharedActions().get(ref.getReference());
if (actionSpec == null) {
throw new IllegalArgumentException(
"Agent '"
+ entry.getKey()
+ "' references shared action '"
+ ref.getReference()
+ "' in "
+ path
+ ", but no shared action with that name is defined at"
+ " the file level.");
}
}
ActionSpec shared = loaded.getSharedActions().get(ref.getReference());
if (shared == null) {
throw new IllegalArgumentException(
"Agent '"
+ entry.getKey()
+ "' references shared action '"
+ ref.getReference()
+ "' in "
+ path
+ ", but no shared action with that name is defined at"
+ " the file level.");
}
addActionToAgent(entry.getValue(), shared);
orderedActions.add(actionSpec);
}

entry.getValue().getActions().clear();
for (ActionSpec actionSpec : orderedActions) {
addActionToAgent(entry.getValue(), actionSpec);
}
}

Expand Down Expand Up @@ -427,7 +431,10 @@ static Function resolveActionFunction(ActionSpec action) {
return resolveFunction(action.getName(), action.getFunction(), language, paramTypes);
}

/** Register an action on the agent with event aliases resolved. */
/**
* Registers an action, replacing trigger conditions that exactly match built-in event aliases
* while leaving expression conditions unchanged.
*/
static void addActionToAgent(Agent agent, ActionSpec action) {
Function fn = resolveActionFunction(action);
String[] triggerConditions =
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,16 +21,18 @@
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.agents.api.yaml.Language;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
* Action referencing a user function plus its trigger conditions.
*
* <p>Each entry in {@code trigger_conditions} is either an event-type name (bare identifier) or a
* future condition-expression form — the runtime classifies the string when it loads the plan.
* <p>Each entry in {@code trigger_conditions} is either an event-type name or a condition
* expression. Entries are classified and validated during {@code AgentPlan} construction.
*/
@JsonIgnoreProperties(ignoreUnknown = false)
public final class ActionSpec {
Expand All @@ -41,15 +43,25 @@ public final class ActionSpec {
private final Language type;

@JsonCreator
public ActionSpec(
private ActionSpec(
@JsonProperty(value = "name", required = true) String name,
@JsonProperty("function") String function,
@JsonProperty(value = "trigger_conditions", required = true)
List<String> triggerConditions,
JsonNode triggerConditionsNode,
@JsonProperty("config") Map<String, Object> config,
@JsonProperty("type") Language type) {
this(name, function, parseTriggerConditions(name, triggerConditionsNode), config, type);
}

public ActionSpec(
String name,
String function,
List<String> triggerConditions,
Map<String, Object> config,
Language type) {
if (triggerConditions == null || triggerConditions.isEmpty()) {
throw new IllegalArgumentException("trigger_conditions must not be empty");
throw new IllegalArgumentException(
"'trigger_conditions' is required and must contain at least one entry");
}
this.name = name;
this.function = function;
Expand All @@ -58,6 +70,32 @@ public ActionSpec(
this.type = type;
}

private static List<String> parseTriggerConditions(
String actionName, JsonNode triggerConditionsNode) {
if (triggerConditionsNode == null || triggerConditionsNode.isNull()) {
return null;
}
if (!triggerConditionsNode.isArray()) {
throw new IllegalArgumentException("'trigger_conditions' must be an array");
}

List<String> triggerConditions = new ArrayList<>(triggerConditionsNode.size());
for (int index = 0; index < triggerConditionsNode.size(); index++) {
JsonNode entry = triggerConditionsNode.get(index);
if (entry == null || entry.isNull()) {
triggerConditions.add(null);
} else if (entry.isTextual()) {
triggerConditions.add(entry.textValue());
} else {
throw new IllegalArgumentException(
String.format(
"'trigger_conditions' entry #%d for action '%s' must be a string, but found %s",
index + 1, actionName, entry.getNodeType()));
}
}
return triggerConditions;
}

public String getName() {
return name;
}
Expand Down
27 changes: 17 additions & 10 deletions api/src/test/java/org/apache/flink/agents/api/EventTypeTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -26,22 +26,29 @@
import org.apache.flink.agents.api.event.ToolResponseEvent;
import org.junit.jupiter.api.Test;

import java.util.Map;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;

/** Tests for {@link EventType}. */
class EventTypeTest {

@Test
void builtInConstantsMatchEventClassConstants() {
assertEquals(InputEvent.EVENT_TYPE, EventType.InputEvent);
assertEquals(OutputEvent.EVENT_TYPE, EventType.OutputEvent);
assertEquals(ChatRequestEvent.EVENT_TYPE, EventType.ChatRequestEvent);
assertEquals(ChatResponseEvent.EVENT_TYPE, EventType.ChatResponseEvent);
assertEquals(ToolRequestEvent.EVENT_TYPE, EventType.ToolRequestEvent);
assertEquals(ToolResponseEvent.EVENT_TYPE, EventType.ToolResponseEvent);
assertEquals(
ContextRetrievalRequestEvent.EVENT_TYPE, EventType.ContextRetrievalRequestEvent);
void allConstantsProvidesAnUnmodifiableNameToValueMap() {
assertEquals(
ContextRetrievalResponseEvent.EVENT_TYPE, EventType.ContextRetrievalResponseEvent);
Map.of(
"InputEvent", InputEvent.EVENT_TYPE,
"OutputEvent", OutputEvent.EVENT_TYPE,
"ChatRequestEvent", ChatRequestEvent.EVENT_TYPE,
"ChatResponseEvent", ChatResponseEvent.EVENT_TYPE,
"ToolRequestEvent", ToolRequestEvent.EVENT_TYPE,
"ToolResponseEvent", ToolResponseEvent.EVENT_TYPE,
"ContextRetrievalRequestEvent", ContextRetrievalRequestEvent.EVENT_TYPE,
"ContextRetrievalResponseEvent", ContextRetrievalResponseEvent.EVENT_TYPE),
EventType.allConstants());
assertThrows(
UnsupportedOperationException.class,
() -> EventType.allConstants().put("custom", "custom"));
}
}
Loading
Loading