GitHub user rosemarYuan edited a discussion: [Discussion] Memory Observation
Capability Design
# Memory Observability Design
## Introduction
This doc introduces Memory Observability for Flink Agents — the ability to
observe how an agent's memory changes as the agent runs.
The framework already provides event-based observability (`EventLogger` /
`EventListener`), but memory reads and writes do not produce events today, so
memory operations are invisible to it.
This doc proposes to close that gap by reusing the existing event-based
observability: the framework records each memory operation as it happens, and
at the action execution boundary converts the records into events, so that all
existing tooling (logging, listening, routing, action triggering) works for
memory out of the box.
The sections below will cover:
* The motivation and design goals.
* The API design: configuration options and the memory event format.
* How events are generated internally, and how the `value` payload is encoded.
* Rebuilding the full Short-Term / Sensory Memory state from the event log.
## 1. Motivation
### Background
Flink Agents provides an event-based observability stack: users can record and
export framework events with `EventLogger` / `EventListener`. Meanwhile, the
framework defines three types of memory with the following operations:
* Sensory Memory / Short-Term Memory (STM): Read / Write
* Long-Term Memory (LTM): Add / Delete / Get / Search
### Problem
Users can observe an agent's run results through events. But for agents that
use memory, there is currently no way to observe how memory evolves inside the
agent.
### Goal
Use the existing event-based observability and turn memory operation into a
memory-event. Concretely: record each memory operation synchronously as it
happens, then converting those records into events at the action boundary.
Existing observability tooling then applies to memory unchanged.
* Keep the existing event observation and routing logic untouched.
* Optionally support per-memory-type, per-operation observation.
```plaintext
// Goal: N memory operations inside one action → a set of subscribable
memory-events
user action : memory operations ──▶ action boundary
[recorded synchronously] │
▼
[converted into N types of memory events]
│
┌────────────────────────────┼────────────────────────────┐
▼ ▼ ▼
EventLogger EventListener routing + action
triggering
```
Target usage:
```java
// Enable memory observability
AgentConfiguration config = new AgentConfiguration();
config.set(MemoryEventOptions.MEMORY_GENERATE_EVENT, true);
// From now on, the memory operations of every action are emitted as
// _<scope>_<operation>_event at the action boundary. For example, the user
// can subscribe to short-term-memory writes of the user.id field:
@Action("type == '_short_term_write_event' && 'user.id' in value")
public void onIdChanged(Event event, RunnerContext ctx) { ... }
```
## 2. API Design
### 2.1 Configuration Options
By default, memory observability generates a set of commonly useful events.
Without any configuration, users can observe the high-value memory operations:
short-term writes, sensory writes, long-term updates, long-term gets, and
long-term searches.
Users who need to adjust the observation scope can control whether each type of
memory operation generates events via `AgentConfiguration`. The framework
provides a master switch and per-operation switches:
* `memory.generate-event`: the master switch. It has no default value; when a
per-operation switch is not configured, the resolution falls back to the master
switch's configured value.
* `memory.generate-event.<scope>-<operation>`: per-operation switches, each
controlling whether one specific kind of memory operation generates events.
| Option key | Default behavior (when neither the per-operation switch nor the
master switch is configured) | Description |
| --- | --- | --- |
| `memory.generate-event` | — (no default value) | Master switch for memory
events. Acts as the fallback when a per-operation switch is not configured; if
the master switch is not configured either, the operation's own default
behavior applies |
| `memory.generate-event.short-term-write` | record | Whether to record
short-term memory writes |
| `memory.generate-event.short-term-read` | do not record | Whether to record
short-term memory reads |
| `memory.generate-event.sensory-write` | record | Whether to record sensory
memory writes |
| `memory.generate-event.sensory-read` | do not record | Whether to record
sensory memory reads |
| `memory.generate-event.long-term-update` | record | Whether to record
long-term memory updates (both writes and deletes) |
| `memory.generate-event.long-term-get` | record | Whether to record long-term
memory gets |
| `memory.generate-event.long-term-search` | record | Whether to record
long-term memory searches |
Per-operation switches take precedence over the master switch. The effective
behavior is resolved in the following order:
1. If the per-operation switch is configured, use its configured value.
2. Otherwise, if the master switch `memory.generate-event` is explicitly
configured, use the master switch's configured value.
3. Otherwise (neither is configured), use the operation's own default behavior
(see the table above).
Usage examples:
```java
// Default behavior (nothing configured): write operations and LTM operations
// are recorded, read operations are not
AgentConfiguration config = new AgentConfiguration();
// Turn off memory observability entirely
config.setBool("memory.generate-event", false);
// Enable only selected operations: turn the master switch off, then observe
// only long-term updates and short-term writes
config.setBool("memory.generate-event", false);
config.setBool("memory.generate-event.long-term-update", true);
config.setBool("memory.generate-event.short-term-write", true);
```
### 2.2 Memory Events
A memory event is a regular event that can be routed and can trigger actions
like any other event. It contains the following fields:
* **id:** the UUID of the event.
* t**ype:** the kind of memory operation, one of:
* `_short_term_write_event`: a write to short-term memory
* `_short_term_read_event`: a read from short-term memory
* `_sensory_write_event`: a write to sensory memory
* `_sensory_read_event`: a read from sensory memory
* `_long_term_update_event`: an update to long-term memory, covering both
add and delete
* `_long_term_get_event`: a get from long-term memory by id
* `_long_term_search_event`: a semantic search over long-term memory
* **key:** the Flink keyed-state key this operation belongs to, identifying
which key the operation happened under.
* **value:** the content of the memory operation. Concretely:
* for write events, the data that was written;
* for read events, the data that was returned.
## 3. Design
### 3.1 Event Generation
At runtime the framework turns every memory operation into an event in two
steps: it records the operation synchronously when it happens, then converts
all records into events at the action boundary.
**Recording — synchronous, at the operation site.** Recording is done inside
the framework's memory operation implementations. The records are written into
the action state and persisted with checkpoints. Users do not notice it, and no
API changes.
**Emission — at the action boundary.** When an action finishes executing, all
memory records produced by that execution are converted into multiple
observation events, grouped by "scope-operation":
```java
{
"id": "<UUID>", // UUID of the event
"type": "_short_term_write_event", // memory operation type
"key": "key1", // the keyed-state key of this operation
"value": { ... } // content of the memory operation
}
{
"id": "<UUID>",
"type": "_long_term_search_event",
"key": "key1",
"value": { ... }
}
```
When an action performs multiple operations of the same kind, the records are
merged by the following rules:
* **Operations on different fields:** coexist in one event, keyed by field:
`{k1: v1, k2: v2, ...}`.
* **Operations on the same field:** only the last value is kept.
In addition, memory operations performed inside an action that was itself
triggered by a memory event do not generate new observation events. This
prevents infinite feedback loops.
### 3.2 The `value` Field
A `MemoryEvent` carries the fields `**id**`**,** `**type**`**,** `**key**`**,
and** `**value**`:
```java
{
"id": "<UUID>", // UUID of the event
"type": "_<scope>_<operation>_event", // memory operation type
"key": "<keyed-state key>", // the Flink keyed-state key of this
operation
"value": { ... } // updates / reads
}
```
For `**value**`, three internal encodings were considered. Take recording an
STM write of `user.tier = "gold"` and `user.address.city = "SF"` as an example:
| **Candidate** | **Shape** | **How a downstream action subscribes to a field**
| **Pros** | **Cons** |
| --- | --- | --- | --- | --- |
| Leaf names | `{"tier":"gold","city":"SF"}` | `has(tier)` | Simple and
intuitive | ❌ Leaf names are not globally unique — ambiguity cannot be resolved
(e.g. `user.id` vs `address.id`) |
| Dotted keys | `{"user.tier":"gold",``"user.address.city":"SF"}` |
`"user.tier" in value` | Unambiguous semantics | Some redundancy: entries
sharing a common prefix are not merged |
| Nested map | `{"user":{` <br>`"tier":"gold",`
<br>`"address":``"city":"SF"}}}` | `has(value.user.tier)` | 1. Structured, more
readable.<br>2. Shared prefixes are merged, saving space | ❌ Cannot tell which
key was operated on — the semantics are ambiguous.E.g. for `{a: {b: 1, c: 2}}`,
the following two operation sequences are indistinguishable:1. `set(a.b, 1);
set(a.c, 2)`2. `set(a, {b: 1, c: 2})` |
Since the nested-map encoding cannot distinguish writing multiple leaf values
field by field from writing a whole map at once, we choose the dotted-key
encoding:
```java
// ① STM write — writes the two leaves user.tier and user.address.city
{ "type": "_short_term_write_event", "id": "<uuid>", "key": "user-42",
"value": { "user.tier": "gold", // add value
"user.address.city": {
"Shanghai": 1201,
"Beijing" : 2021 } // add map
} }
// ② STM read — reads user.address.city
{ "type": "_short_term_read_event", "id": "<uuid>", "key": "user-42",
"value": { "user.address.city": "SF" } }
// ③ LTM update — key = "memorySet.itemId", value = content; null = delete
{ "type": "_long_term_update_event", "id": "<uuid>", "key": "user-42",
"value": {
"user_prefs.m_8f3": "user prefers email notifications", // add
"profile.m_a01": null // delete
} }
// ④ LTM get (by id)
{ "type": "_long_term_get_event", "id": "<uuid>", "key": "user-42",
"value": {
"user_prefs.m_8f3": "user prefers email notifications",
"user_prefs.m_a01": "user is located in SF"
} }
// ⑤ LTM search — not dotted-key encoded; keeps the query → ordered result
list form
{ "type": "_long_term_search_event", "id": "<uuid>", "key": "user-42",
"value": {
"refund policy": [
{ "id": "p_01", "value": "7-day no-questions-asked refund", "score": 0.92
},
{ "id": "p_02", "value": "customized items are non-refundable", "score":
0.81 }
]
} }
```
## 4. Rebuilding Short-Term / Sensory Memory State from the Event Log
Besides observing memory at runtime and triggering new actions, there is a
second class of demand: during log-based troubleshooting, reconstructing the
memory state at a given moment from the event log.
**The basic idea:** since the framework records every memory update from job
start, applying the updates in order on top of a known initial state restores
the state at any given moment. Feasibility differs per memory type, though:
* **LTM:** user-written data goes through LLM processing (compaction,
merging, deletion), so the state cannot be rebuilt by applying updates to an
initial state.
* **Sensory Memory:** automatically cleared at the end of every agent run, so
applying the `write` records of the current run is sufficient.
* **STM:** its content keeps evolving for the lifetime of the job. Replaying
from job start makes the rebuild cost grow linearly with job age, so periodic
full snapshots are needed as rebuild starting points.
The rest of this section discusses when to record the full snapshots and how
the rebuild works.
### 4.1 When to Record Full Memory Snapshots
The rebuild needs periodic full STM snapshots as starting points, i.e.
recording the complete STM state per key every so often.
**Chosen: record at agent-run begin.** Whenever an input for some key arrives
and an agent run is about to start, record a full STM snapshot for that key.
Concretely, when an `InputEvent` arrives, the framework emits an
`AgentRunBeginEvent` marking the start of the agent run. The event carries the
full STM as of the input's arrival, to be used for state reconstruction from
the event log. The memory payload format is the same as the `value` structure
in §3.2.
**Rejected alternative:** record at checkpoint time. On checkpoint completion
(`notifyCheckpointComplete`), iterate over all keys held by the subtask via
`applyToAllKeys` and record a full STM snapshot per key. This was rejected
because the checkpointing path already carries a lot of responsibilities.
**Usage:**
```java
// Enable AgentRunBeginEvent
config.set(MemoryEventOptions.AGENT_RUN_BEGIN_EVENT, true);
```
**Configuration:**
```java
// AgentRunBeginEvent is a run-lifecycle event. It is not governed by the
// memory event master switch, and is recorded by default.
public static final ConfigOption<Boolean> AGENT_RUN_BEGIN_EVENT =
new ConfigOption<>("agent-run.begin-event", Boolean.class, true);
```
### 4.2 How the Rebuild Works
Since every agent run starts with a full STM snapshot, the rebuild needs
neither a standalone command-line tool nor replaying long changelogs merged
across runs. To rebuild the state of a key at time t:
1. Locate, in the event log, the latest full snapshot of that key before t
(i.e. the `AgentRunBeginEvent` of the run containing t).
2. Starting from that snapshot, apply the `_short_term_write_event`s of that
run before t in order — this yields the state at t. If t is exactly a run
boundary, the snapshot itself is the answer.
The rebuild cost is proportional to the number of operations within a single
run and does not grow with job age.
Likewise, since Sensory Memory always starts empty, applying the `write`
records of the current run reconstructs its state at any moment.
## Future Works
* A safety valve for large `value` payloads (e.g. size limits / truncation).
* Optionally recording the previous value alongside the new value for richer
diffs.
* Recording full snapshots every N runs (counter-based) instead of every run,
to reduce snapshot overhead.
* An `AgentRunEndEvent` complementing `AgentRunBeginEvent`.
GitHub link: https://github.com/apache/flink-agents/discussions/876
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
