You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "rosemarYuan (via GitHub)" <gi...@apache.org> on 2026/07/07 02:46:35 UTC
Re: [D] [Discussion] Memory Oobservation Capability Design [flink-agents]
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 issues@flink.apache.org.
To unsubscribe, please send an email to: issues-unsubscribe@flink.apache.org