Memory System Documentation

Скачать · Исходный код · О проекте

Memory System Documentation

Overview

The memory system has two complementary layers:

Multi-level runtime memory (MemoryManager + WorkingMemoryCompressor) — L0–L3 hierarchy under {runtime_data_dir}/memory/ (default ~/.local-ai/memory/: state, working summary, plans, tool results). Requires local-ai-core[memory] and memory.enabled: true. L0/L2 wired in AgentLoop; L1 when memory.working_memory_enabled; L3 via ToolExecutor + load_tool_result.
Indexed RAG memory (MemoryStore + MemoryRetriever) — Markdown knowledge in {runtime_data_dir}/memory_data/. Requires local-ai-core[memory] and memory.enabled: true. Categories from memory.categories (global, not per-agent).

Multi-Level Memory (`packages/local-ai-memory/memory/manager.py`)

MemoryManager provides a file-backed hierarchy for agent runtime state, separate from the category-based memory_data/ RAG pipeline.

Storage layout

Paths are scoped per {user_id}_{session_id} when SessionRunner sets memory scope (auto-migrates global files on first use):

{runtime_data_dir}/memory/   # default ~/.local-ai/memory/
├── state/{user_id}_{session_id}.json   # L0 — agent state
├── working/{user_id}_{session_id}.md   # L1 — working summary
├── plans/{user_id}_{session_id}.json   # L2 — checkpoints (AgentCheckpoint)
├── tool_results/{result_id}.json       # L3 — externalized tool results

Levels

Level	API	Default on read
L0 State	`get_state()`, `update_state(patch)`	`{}`
L1 Working	`get_working_memory()`, `update_working_memory(content)`	`""`
L2 Plans	`save_plan(id, payload)`, `load_plan(id)`	`FileNotFoundError`
L3 Tool results	`save_tool_result(id, payload)`, `load_tool_result(id)`	`FileNotFoundError`

Versioned envelopes

JSON files use format_version: 1 envelopes. L0 example:

{
  "format_version": 1,
  "updated_at": 1718123456.0,
  "state": { "current_task": "...", "current_step": 2 }
}

L2/L3 wrap payloads as { "format_version", "id", "saved_at", "payload" }.

L0 agent state and L2 checkpoints

L0 is read into context as formatted agent_state text and updated each ReAct step (current_task, current_step, plan_id).

L2 stores AgentCheckpoint envelopes in plans/{user_id}_{session_id}.json:

Status	Meaning
`in_progress`	Run interrupted or mid-loop; resumable
`completed`	Run finished normally
`interrupted`	SIGINT/SIGTERM graceful shutdown

AgentLoop saves a checkpoint after each tool step. Resume with CLI --resume or session_runner.run(..., resume=True).

L3 tool result externalization

Large successful tool outputs are kept out of LLM context by tools/tool_result_store.py, called from ToolExecutor after execution.

Threshold: app.tool_result_inline_limit (default 6144 UTF-8 bytes). Results at or below the limit are returned inline unchanged.

Flow:

Tool.func → ToolExecutor.execute()
  → size <= limit? → full content to agent
  → size > limit?  → MemoryManager.save_tool_result()
                   → agent receives JSON reference

Stored payload (inside the L3 envelope payload field):

{
  "id": "a1b2c3d4e5f6...",
  "tool": "mcp_filesystem_read_file",
  "created_at": "2026-06-11T12:00:00+00:00",
  "content": "<full tool output>"
}

Agent-facing reference (replaces ToolResult.content in observations):

{
  "result_id": "a1b2c3d4e5f6...",
  "summary": "<first N characters>…",
  "size": 128450
}

Only successful results are externalized; error messages always stay inline.
summary length is app.tool_result_summary_max_chars (default 1000, with … when truncated); size is the original UTF-8 byte count.
Log event: TOOL_RESULT_STORED with tool, result_id, and size.
Full content via MemoryManager.load_tool_result(result_id) or agent tool load_tool_result (from local_ai_memory.tools.create_memory_tools when memory is wired).

Persistence guarantees

Atomic writes — .tmp file + Path.replace() (same pattern as SessionStore)
Thread safety — per-file asyncio.Lock; full RMW under lock
Corruption recovery — backup to {name}.corrupt.bak, log MEMORY_CORRUPT, return safe defaults
ID sanitization — plan/result IDs must match [A-Za-z0-9_-]+ (blocks path traversal)

L1 working memory compression (`packages/local-ai-memory/memory/compressor.py`)

WorkingMemoryCompressor LLM-summarizes conversation state into L1 working_memory.md, keeping prompts compact while preserving task context across turns.

Structured sections (markdown):

# Current Goal
...

# Completed
* ...

# Important Discoveries
* ...

# Open Issues
* ...

# Decisions
* ...

API:

Method	Purpose
`compress(messages)`	Full rebuild from a conversation transcript
`update_summary(new_messages)`	Incremental merge with existing L1

Runtime flow (wired in app.py → AgentLoop):

AgentLoop.run()
  ├─ read L1 → ContextBuilder ("Working memory:" section)
  ├─ ReAct loop (RAG memory + session history + tools)
  └─ post-run: update_summary(run_messages)
       └─ if session > history_limit → SessionStore.compact(keep_last)

Configuration (memory in config.yaml):

Key	Default	Purpose
`working_memory_target_tokens`	`768`	Max tokens for L1 content
`working_memory_enabled`	`true`	Toggle L1 read/compress in `AgentLoop`
`working_memory_compact_keep_last`	`6`	Session messages kept after post-run compact

Budget enforcement: TokenCounter checks output size; over-budget summaries trigger an LLM shrink pass, then deterministic section trimming.

L0 sync: when # Current Goal is non-empty, current_task is written to L0 state.

Log event: WORKING_MEMORY_COMPRESSED with mode, tokens_before, tokens_after.

Failure behavior: on LLM error, existing L1 is left unchanged (no exception leakage to the agent).

Context trim priority (ContextBuilder): oldest history first → RAG memory → working memory last.

Usage

from memory.manager import MemoryManager
from security.guard import DirectoryGuard

manager = MemoryManager(DirectoryGuard(workspace_root))

await manager.update_state({"current_task": "refactor auth", "current_step": 1})

# L1 is normally maintained by WorkingMemoryCompressor at end of each AgentLoop run:
from memory.compressor import WorkingMemoryCompressor

compressor = WorkingMemoryCompressor(manager, llm_client, token_counter, target_tokens=768)
await compressor.update_summary([
    {"role": "user", "content": "Refactor the auth module."},
    {"role": "assistant", "content": "I'll start by reading the current auth code."},
])

# Direct L1 write (bypasses compressor):
await manager.update_working_memory("# Working\n\nFocused on login flow.")

await manager.save_plan("auth-refactor", {"steps": ["read", "write", "test"]})
plan = await manager.load_plan("auth-refactor")

await manager.save_tool_result("abc123", {
    "id": "abc123",
    "tool": "mcp_filesystem_read_file",
    "created_at": "2026-06-11T12:00:00+00:00",
    "content": "file contents…",
})
stored = await manager.load_tool_result("abc123")

Run tests:

python -m pytest packages/local-ai-memory/tests/test_memory_manager.py -v — L0–L3 APIs
python -m pytest packages/local-ai-memory/tests/test_memory_compressor.py -v — L1 compression and token budget
python -m pytest packages/local-ai-memory/tests/test_tool_result_store.py -v — large output externalization via ToolExecutor

Indexed RAG Memory (`MemoryStore` + `MemoryRetriever`)

Purpose

Store knowledge - Persist agent knowledge in organized Markdown files
Retrieve relevant context - Find relevant information based on queries
Manage token budget - Load only necessary content within token limits
Scale efficiently - Index-based pre-filtering reduces LLM calls from O(n) to O(1)

Responsibilities

Store management - Create, read, update, list memory files
Indexing - Build and maintain a persistent memory_index.json with metadata
Auto-refresh - Detect stale files via mtime and re-index without LLM
Retrieval - 3-stage retrieval (BM25 or hybrid → LLM select → budgeted load)
User separation - Isolate memory by user_id
Category organization - Organize by knowledge categories

Dependencies

security/guard.py - DirectoryGuard for sandbox enforcement
core/llm_client.py - LocalLLMClient for LLM-based indexing (full population only)
core/token_counter.py - TokenCounter for budget management
asyncio - Async file operations

Constraints

Sandbox enforcement - All paths go through DirectoryGuard
Token budget - Content loading respects max_tokens parameter
TOP_N - Stage 1 returns top 5 candidates (TOP_N = 5 in retriever.py)
Selection limit - Stage 2 selects max 5 files
Search mode - Default BM25 (memory.search_mode: bm25); optional hybrid BM25+embeddings with RRF fusion (hybrid)

Extension Points

Adding a New Memory Type

The current system uses Markdown files. To add a new type:

Create a new MemoryStore subclass
Implement list_user_files(), read_file(), bm25_search() / hybrid_search()
Update MemoryRetriever to support the new type
Ensure DirectoryGuard compliance

Adding Categories

Categories are configured globally in config.yaml:

memory:
  enabled: true
  categories: ["knowledge", "personal", "work"]

Each category creates a subdirectory: {runtime_data_dir}/memory_data/{category}/

Module Details

`memory/store.py` - MemoryStore

Manages Markdown memory files and the persistent memory_index.json. All file access is sandboxed through DirectoryGuard.

Core Methods:

list_user_files(user_id, category) - List all user files in category
read_file(path) / write_file(path, content) - Async file I/O via asyncio.to_thread
category_dir(category) - Resolve and create a category subdirectory

Index Management Methods:

index_exists() - Check if memory_index.json exists
read_index() - Read index from disk
write_index(index) - Write index to disk (atomically via .tmp rename)
update_index_entry(filename, title, description, keywords, category) - Incremental update of a single entry (no full rebuild)
populate_index_from_files(user_id, categories, llm_client) - Full index build using LLM for keyword enrichment
bm25_search(query, top_n=5) - BM25 ranking via BM25Ranker + extract_query_keywords()
fast_filter(query, top_n=5) - Alias for bm25_search()
hybrid_search(query, top_n, bm25_weight) - BM25 + embedding vectors, RRF fusion (memory/hybrid_search.py)
get_stale_entries(user_id, categories) - Detect files whose mtime differs from indexed last_modified
refresh_stale_entries(user_id, categories) - Re-index stale files using heuristic keywords (no LLM)
_extract_keywords_heuristic(title, description, full_content) - Keyword extraction without LLM

Index structure (v3):

{
  "index_version": 3,
  "last_updated": 1686234567.123,
  "files": [
    {
      "filename": "user1_notes.md",
      "title": "Personal Notes",
      "description": "Collection of personal notes and reminders about programming",
      "keywords": ["notes", "personal", "reminders", "programming"],
      "category": "knowledge",
      "last_modified": 1686234567.0
    }
  ]
}

Index Population (full):

Scan all files for title (H1), description (first paragraph), and initial heuristic keywords from full content
Use LLM to generate 5-10 refined keywords per file
Store complete index as memory_index.json

Index Auto-Refresh (incremental):

On each retrieve() call, compare each file's stat().st_mtime against indexed last_modified
Stale or new files are re-scanned for metadata
Keywords are extracted via _extract_keywords_heuristic() — no LLM round-trip
Index is updated incrementally via update_index_entry()

Only the initial population (index missing) uses the LLM. Subsequent auto-refreshes are LLM-free.

`memory/index_service.py` — MemoryIndexService

Builds/refreshes index on ctx.startup()
Polls memory_data/ every 3 seconds for create/modify/delete
Rebuilds when index_version < current (_INDEX_VERSION = 3)

`memory/retriever.py` - MemoryRetriever

Implements 3-stage indexed retrieval:

Flow:

Query
  │
  ├─ [Index missing/stale?] → MemoryIndexService / refresh_stale_entries
  │
  ▼
Stage 1: bm25_search (default) OR hybrid_search (memory.search_mode: hybrid)
  │  NO LLM — returns top 5 (TOP_N)
  │
  ▼
Stage 2: LLM selects from ≤5 candidates (rich metadata)
  │
  ▼
Stage 3: Load selected files within token budget
  │
  ▼
Return concatenated content

Stage 1: BM25 / Hybrid filtering

BM25 (default): bm25_search() ranks index entries via BM25Ranker and extract_query_keywords(). fast_filter() is an alias.
Hybrid: when memory.search_mode: hybrid, hybrid_search() combines BM25 results with embedding similarity (llm.embed()), fused via reciprocal rank fusion (RRF). Embeddings stored under memory_data/.embeddings/. Falls back to BM25 if embeddings unavailable.
Config: memory.hybrid_bm25_weight (default 0.5), hybrid_top_n (default 10, used as Stage 1 pool size in hybrid mode).

Stage 2: LLM File Selection

Receives rich metadata for each candidate (title, description, keywords, category)
LLM selects the most relevant files (max 5)
Returns strict JSON array of filenames
If parsing fails → empty list (no crash)

Stage 3: Content Loading

Uses index to resolve category for each selected file (avoids full directory scan)
Loads files in order, tracking token count via TokenCounter
Stops when max_tokens budget is reached
Returns concatenated ## filename\n{content} blocks

Retrieval Algorithm

Pre-check: Stale Index Detection

On every retrieve() call, after confirming the index exists:

stale = []
for category in categories:
    for path in list_user_files(user_id, category):
        current_mtime = path.stat().st_mtime          # stat() only, no read
        stored_mtime = indexed.get((category, path.name), 0.0)
        if current_mtime > stored_mtime:
            stale.append((category, path))

for category, path in stale:
    metadata = scan_file_metadata(path, category)     # reads file content
    update_index_entry(
        filename=metadata["filename"],
        title=metadata["title"],
        description=metadata["description"],
        keywords=metadata["keywords"],               # heuristic, no LLM
        category=category,
    )

Stage 1: BM25 search

async def bm25_search(query: str, top_n: int = 5) -> list[str]:
    index = await read_index()
    if not index:
        return []
    keywords = extract_query_keywords(query)
    ranker = BM25Ranker(index["files"])
    return ranker.rank(keywords, top_n=top_n)

fast_filter() delegates to bm25_search(). Hybrid mode uses hybrid_search() with RRF over BM25 and vector results.

Stage 2: LLM Selection (Rich Metadata)

Prompt (with enriched candidate info):

You are a memory file selector.
Given a user query and available file structure metadata, return ONLY a JSON array of filenames to load.

QUERY:
{python programming basics}

CANDIDATE FILES:
File: user1_python.md
  Title: Python Basics
  Category: knowledge
  Description: Introduction to Python programming
  Keywords: python, programming, tutorial, basics
---
File: user1_rust.md
  Title: Rust Programming
  Category: knowledge
  Description: Learning Rust and systems programming
  Keywords: rust, programming, systems, concurrency

Select the most relevant files (up to 5). Return ONLY a JSON array of filenames.

Response Parsing:

Parse JSON array from LLM response
Validate all items are strings
Limit to TOP_N (default 5)
Return empty list if parsing fails (graceful degradation)

Stage 3: Content Loading (Index-Aware)

async def load_selected(user_id, categories, selected, max_tokens):
    # Read index to resolve category per file
    index = read_index()
    filename_to_cat = {e["filename"]: e.get("category", "") for e in index}
    
    chunks, used_tokens = [], 0
    for name in selected:
        cat = filename_to_cat.get(name, "")
        if cat:
            path = category_dir(cat) / name          # direct path from index
        else:
            path = find_in_categories(categories, name)  # fallback scan
        
        content = read_file(path)
        tokens = count_tokens(content)
        if used_tokens + tokens > max_tokens:
            break
        chunks.append(f"## {name}\n{content}")
        used_tokens += tokens
    
    return "\n\n".join(chunks)

Usage Examples

Basic Retrieval (with auto-refresh)

from memory.retriever import MemoryRetriever
from memory.store import MemoryStore

memory_store = MemoryStore(guard)
retriever = MemoryRetriever(memory_store, llm_client, token_counter)

# Retrieval automatically:
# 1. Checks if index exists → populates if missing
# 2. Detects stale files via mtime → refreshes if needed (no LLM)
# 3. Runs Stage 1 bm25_search or hybrid_search (no LLM, returns top 5)
# 4. Runs Stage 2 LLM selection on ≤5 candidates
# 5. Loads selected files within budget
memory = await retriever.retrieve(
    query="python programming basics",
    user_id="default_user",
    categories=["knowledge"],
    max_tokens=2000,
)

Full Index Population (LLM-assisted)

# Automatically triggered on first retrieve() when index is missing
memory = await retriever.retrieve(...)

# Or manually:
await memory_store.populate_index_from_files(
    user_id="default_user",
    categories=["knowledge"],
    llm_client=llm_client,  # LLM generates keywords per file
)

Incremental Index Update (no LLM)

# Add a new file manually
path = memory_store.category_dir("knowledge") / "default_user_new.md"
await memory_store.write_file(path, "# New Topic\nImportant content here")

# Trigger auto-refresh via normal retrieval
await retriever.retrieve(query="new topic", ...)
# → get_stale_entries() detects new file by mtime
# → refresh_stale_entries() re-indexes with heuristic keywords (no LLM)

Check Staleness Manually

stale = await memory_store.get_stale_entries(
    user_id="default_user",
    categories=["knowledge"],
)
print(f"{len(stale)} stale files need re-indexing")

File Organization

workspace/
├── memory/                        ← Multi-level runtime memory (MemoryManager)
│   ├── state/                     ← scoped L0 JSON
│   ├── working/                   ← scoped L1 markdown
│   ├── plans/                     ← scoped L2 checkpoints
│   ├── tool_results/              ← L3 externalized outputs
└── memory_data/                   ← Indexed RAG knowledge (MemoryStore)
    ├── memory_index.json          ← single index (index_version: 3)
    ├── .embeddings/               ← hybrid search vectors (when enabled)
    ├── knowledge/
    │   └── default_user_*.md
    └── {category}/                ← additional RAG categories from memory.categories

Note: memory_data/memory_index.json covers all configured RAG categories in one file.

Security

DirectoryGuard - All paths resolved through guard
User isolation - Files named with user_id prefix
Category sandbox - Each category is separate directory
No external access - Only reads from memory_data/

Performance

Index check / Auto-refresh: Only stat() calls (O(n) on mtime comparison, no content reads except for stale files)
Stage 1: BM25 or hybrid rank over index — no LLM call
Stage 2: One LLM call with ≤5 candidates
Stage 3: Only loads necessary content within budget

Optimization:

Index provides fast filtering with no LLM
LLM only sees top 5 candidates (not all files)
Auto-refresh is LLM-free for changed files
Token budget prevents overloading context

Testing

Key test cases:

Empty index → auto-populate with LLM
Existing index → stale detection via mtime / MemoryIndexService watcher
BM25 and hybrid Stage 1 ranking
Token budget enforcement in content loading
Invalid LLM JSON → graceful fallback to empty
Incremental update_index_entry vs full populate_index_from_files

Run:

python -m pytest packages/local-ai-memory/tests/test_memory_manager.py -v — multi-level MemoryManager
python -m pytest packages/local-ai-memory/tests/test_memory_compressor.py -v — L1 working memory compression
python -m pytest packages/local-ai-memory/tests/test_tool_result_store.py -v — L3 tool result externalization
python -m pytest packages/local-ai-memory/tests/test_memory_retriever.py -v — indexed RAG retriever

Architecture Rules for Memory

Sandbox enforcement - All file access through DirectoryGuard
User isolation - Memory files prefixed with user_id
Index first - Always use index for fast filtering before LLM
Auto-refresh - MemoryIndexService watcher + mtime-based refresh_stale_entries()
No LLM in Stage 1 - BM25/hybrid filtering is LLM-free
Token budget - Never exceed max_tokens in content loading
Lazy loading - Don't load all files, filter first via index
LLM as final selector - LLM only sees ≤5 candidates for refinement

Memory System Documentation

Memory System Documentation

Overview

Multi-Level Memory (packages/local-ai-memory/memory/manager.py)

Storage layout

Levels

Versioned envelopes

L0 agent state and L2 checkpoints

L3 tool result externalization

Persistence guarantees

L1 working memory compression (packages/local-ai-memory/memory/compressor.py)

Usage

Indexed RAG Memory (MemoryStore + MemoryRetriever)

Purpose

Responsibilities

Dependencies

Constraints

Extension Points

Adding a New Memory Type

Adding Categories

Module Details

memory/store.py - MemoryStore

memory/index_service.py — MemoryIndexService

memory/retriever.py - MemoryRetriever

Retrieval Algorithm

Pre-check: Stale Index Detection

Stage 1: BM25 search

Stage 2: LLM Selection (Rich Metadata)

Stage 3: Content Loading (Index-Aware)

Usage Examples

Basic Retrieval (with auto-refresh)

Full Index Population (LLM-assisted)

Incremental Index Update (no LLM)

Check Staleness Manually

File Organization

Security

Performance

Testing

Architecture Rules for Memory

Навигация

Информация

Сообщество

Multi-Level Memory (`packages/local-ai-memory/memory/manager.py`)

L1 working memory compression (`packages/local-ai-memory/memory/compressor.py`)

Indexed RAG Memory (`MemoryStore` + `MemoryRetriever`)

`memory/store.py` - MemoryStore

`memory/index_service.py` — MemoryIndexService

`memory/retriever.py` - MemoryRetriever