GitHub user rosemarYuan edited a discussion: [Discussion]Memory Oobservation
Capability Design
<html><head></head><body><h2>Introduction</h2>
<p>This doc introduces Memory Observability for Flink Agents — the ability to
observe how an agent's memory changes as the agent runs.</p>
<p>The framework already provides event based observability
(<code>EventLogger</code> / <code>EventListener</code>), but memory reads and
writes do not produce events today, so memory operations are invisible to
it.</p>
<p>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.</p>
<p>The sections below will cover:</p>
<ul>
<li>The motivation and design goals.</li>
<li>The API design: configuration options and the memory event format.</li>
<li>How events are generated internally, and how the <code>value</code> payload
is encoded.</li>
<li>Rebuilding the full Short-Term / Sensory Memory state from the event
log.</li>
</ul>
<h2>1. Motivation</h2>
<h3>Background</h3>
<p>Flink Agents provides an event-based observability stack: users can record
and export framework events with <code>EventLogger</code> /
<code>EventListener</code>. Meanwhile, the framework defines three types of
memory with the following operations:</p>
<ul>
<li>Sensory Memory / Short-Term Memory (STM): Read / Write</li>
<li>Long-Term Memory (LTM): Add / Delete / Get / Search</li>
</ul>
<h3>Problem</h3>
<p>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.</p>
<h3>Goal</h3>
<p>Use the existing event-based observability and turn memory operations into
memory-events. Concretely: record each memory operation synchronously as it
happens, then convert those records into events at the action boundary.
Existing observability tooling then applies to memory unchanged.</p>
<ul>
<li>Keep the existing event observation and routing logic untouched.</li>
<li>Optionally support per-memory-type, per-operation observation.</li>
</ul>
<pre><code class="language-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
</code></pre>
<p>Target usage:</p>
<pre><code class="language-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) { ... }
</code></pre>
<h2>2. API Design</h2>
<h3>2.1 Configuration Options</h3>
<p>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.</p>
<p>Users who need to adjust the observation scope can control whether each type
of memory operation generates events via <code>AgentConfiguration</code>. The
framework provides a master switch and per-operation switches:</p>
<ul>
<li><code>memory.generate-event</code>: 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.</li>
<li><code>memory.generate-event.<scope>-<operation></code>:
per-operation switches, each controlling whether one specific kind of memory
operation generates events.</li>
</ul>
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
<p>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:</p>
<pre><code class="language-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 }
]
} }
</code></pre>
<h2>4. Rebuilding Short-Term / Sensory Memory State from the Event Log</h2>
<p>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.</p>
<p><strong>The basic idea:</strong> 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:</p>
<ul>
<li><strong>LTM:</strong> user-written data goes through LLM processing
(compaction, merging, deletion), so the state cannot be rebuilt by applying
updates to an initial state.</li>
<li><strong>Sensory Memory:</strong> automatically cleared at the end of every
agent run, so applying the <code>write</code> records of the current run is
sufficient.</li>
<li><strong>STM:</strong> 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.</li>
</ul>
<p>The rest of this section discusses when to record the full snapshots and how
the rebuild works.</p>
<h3>4.1 When to Record Full Memory Snapshots</h3>
<p>The rebuild needs periodic full STM snapshots as starting points, i.e.
recording the complete STM state per key every so often.</p>
<p><strong>Chosen: record at agent-run begin.</strong> 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 <code>InputEvent</code> arrives, the
framework emits an <code>AgentRunBeginEvent</code> 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 <code>value</code> structure in §3.2.</p>
<p><strong>Rejected alternative:</strong> record at checkpoint time. On
checkpoint completion (<code>notifyCheckpointComplete</code>), iterate over all
keys held by the subtask via <code>applyToAllKeys</code> and record a full STM
snapshot per key. This was rejected because the checkpointing path already
carries a lot of responsibilities.</p>
<p><strong>Usage:</strong></p>
<pre><code class="language-plaintext">// Enable AgentRunBeginEvent
config.set(MemoryEventOptions.AGENT_RUN_BEGIN_EVENT, true);
</code></pre>
<p><strong>Configuration:</strong></p>
<pre><code class="language-plaintext">// 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);
</code></pre>
<h3>4.2 How the Rebuild Works</h3>
<p>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:</p>
<ol>
<li>Locate, in the event log, the latest full snapshot of that key before t
(i.e. the <code>AgentRunBeginEvent</code> of the run containing t).</li>
<li>Starting from that snapshot, apply the
<code>_short_term_write_event</code>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.</li>
</ol>
<p>The rebuild cost is proportional to the number of operations within a single
run and does not grow with job age.</p>
<p>Likewise, since Sensory Memory always starts empty, applying the
<code>write</code> records of the current run reconstructs its state at any
moment.</p>
<h2>Future Works</h2>
<ul>
<li>A safety valve for large <code>value</code> payloads (e.g. size limits /
truncation).</li>
<li>Optionally recording the previous value alongside the new value for richer
diffs.</li>
<li>Recording full snapshots every N runs (counter-based) instead of every run,
to reduce snapshot overhead.</li>
<li>An <code>AgentRunEndEvent</code> complementing
<code>AgentRunBeginEvent</code>.</li>
</ul></body></html>
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]
