[api][runtime] Support use long-term memory in action and async compaction.

api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java

@@ -43,4 +43,8 @@ public class AgentConfigOptions {
     /** The config parameter specifies the replication factor for the Kafka action state topic. */
     public static final ConfigOption<Integer> KAFKA_ACTION_STATE_TOPIC_REPLICATION_FACTOR =
             new ConfigOption<>("kafkaActionStateTopicReplicationFactor", Integer.class, 1);
+
+    /** The config parameter specifies the unique identifier of job. */
+    public static final ConfigOption<String> JOB_IDENTIFIER =
+            new ConfigOption<>("job-identifier", String.class, null);
 }

python/flink_agents/api/core_options.py

@@ -49,7 +49,10 @@ def covert_j_option_to_python_option(j_option: Any) -> ConfigOption:
 
 class AgentConfigOptionsMeta(type):
     """Metaclass for FlinkAgentsCoreOptions."""
-    def __init__(cls, name: str, bases: tuple[type, ...], attrs: dict[str, Any]) -> None:
+
+    def __init__(
+        cls, name: str, bases: tuple[type, ...], attrs: dict[str, Any]
+    ) -> None:
         """Initialize the metaclass for FlinkAgentsCoreOptions."""
         super().__init__(name, bases, attrs)
 
@@ -68,3 +71,9 @@ def __getattr__(cls, item: str) -> ConfigOption:
 
 class AgentConfigOptions(metaclass=AgentConfigOptionsMeta):
     """CoreOptions to manage core configuration parameters for Flink Agents."""
+
+    JOB_IDENTIFIER = ConfigOption(
+        key="job-identifier",
+        config_type=str,
+        default=None,
+    )

python/flink_agents/api/memory/long_term_memory.py

@@ -30,6 +30,7 @@
 from typing_extensions import override
 
 from flink_agents.api.chat_message import ChatMessage
+from flink_agents.api.configuration import ConfigOption
 from flink_agents.api.prompts.prompt import Prompt
 
 ItemType = str | ChatMessage
@@ -76,6 +77,28 @@ class LongTermMemoryBackend(Enum):
     EXTERNAL_VECTOR_STORE = "external_vector_store"
 
 
+class LongTermMemoryOptions:
+    """Config options for ReActAgent."""
+
+    BACKEND = ConfigOption(
+        key="long-term-memory.",
+        config_type=LongTermMemoryBackend,
+        default=None,
+    )
+
+    EXTERNAL_VECTOR_STORE_NAME = ConfigOption(
+        key="long-term-memory.external-vector-store-name",
+        config_type=str,
+        default=None,
+    )
+
+    ASYNC_COMPACTION = ConfigOption(
+        key="long-term-memory.async-compaction",
+        config_type=bool,
+        default=False,
+    )
+
+
 class DatetimeRange(BaseModel):
     """Represents a datetime range."""
 
@@ -159,7 +182,7 @@ def size(self) -> int:
 
     def add(
         self, items: ItemType | List[ItemType], ids: str | List[str] | None = None
-    ) -> None:
+    ) -> List[str]:
         """Add a memory item to the set, currently only support item with
         type str or ChatMessage.
 
@@ -169,8 +192,11 @@ def add(
         Args:
             items: The items to be inserted to this set.
             ids: The ids of the items to be inserted. Optional.
+
+        Returns:
+            The IDs of the items added.
         """
-        self.ltm.add(memory_set=self, memory_items=items, ids=ids)
+        return self.ltm.add(memory_set=self, memory_items=items, ids=ids)
 
     def get(
         self, ids: str | List[str] | None = None
@@ -203,7 +229,7 @@ class BaseLongTermMemory(ABC, BaseModel):
     def get_or_create_memory_set(
         self,
         name: str,
-        item_type: str | Type[ChatMessage],
+        item_type: type[str] | Type[ChatMessage],
         capacity: int,
         compaction_strategy: CompactionStrategy,
     ) -> MemorySet:
@@ -257,7 +283,7 @@ def add(
         memory_items: ItemType | List[ItemType],
         ids: str | List[str] | None = None,
         metadatas: Dict[str, Any] | List[Dict[str, Any]] | None = None,
-    ) -> None:
+    ) -> List[str]:
         """Add items to the memory set, currently only support items with
         type str or ChatMessage.
 
@@ -269,6 +295,9 @@ def add(
             ids: The IDs of items. Will be automatically generated if not provided.
             Optional.
             metadatas: The metadata for items. Optional.
+
+        Returns:
+            The IDs of added items.
         """
 
     @abstractmethod

python/flink_agents/api/runner_context.py

@@ -16,10 +16,11 @@
 # limitations under the License.
 #################################################################################
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Any, Callable, Dict, Tuple
+from typing import TYPE_CHECKING, Any, Callable, Dict
 
 from flink_agents.api.configuration import ReadableConfiguration
 from flink_agents.api.events.event import Event
+from flink_agents.api.memory.long_term_memory import BaseLongTermMemory
 from flink_agents.api.metric_group import MetricGroup
 from flink_agents.api.resource import Resource, ResourceType
 
@@ -107,6 +108,17 @@ def short_term_memory(self) -> "MemoryObject":
           The root object of the short-term memory.
         """
 
+    @property
+    @abstractmethod
+    def long_term_memory(self) -> BaseLongTermMemory:
+        """Get the long-term memory.
+
+        Returns:
+        -------
+        BaseLongTermMemory
+          The long-term memory instance.
+        """
+
     @property
     @abstractmethod
     def agent_metric_group(self) -> MetricGroup:
@@ -133,8 +145,8 @@ def action_metric_group(self) -> MetricGroup:
     def execute_async(
         self,
         func: Callable[[Any], Any],
-        *args: Tuple[Any, ...],
-        **kwargs: Dict[str, Any],
+        *args: Any,
+        **kwargs: Any,
     ) -> Any:
         """Asynchronously execute the provided function. Access to memory
          is prohibited within the function.
@@ -143,9 +155,9 @@ def execute_async(
         ----------
         func : Callable
             The function need to be asynchronously processing.
-        *args : tuple
+        *args : Any
             Positional arguments to pass to the function.
-        **kwargs : dict
+        **kwargs : Any
             Keyword arguments to pass to the function.
 
         Returns: