Вернуться обратно

API Reference: Memory & Security

API Reference: Memory & Security

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

API Reference: Memory & Security

security/guard.py - DirectoryGuard

Purpose: Path sandbox for Core-owned I/O (runtime_data_dir, and optional workspace for skills)

DirectoryGuard(base_path: Path)

class DirectoryGuard:
    def __init__(self, base_path: Path)
    
    def resolve(self, path: str) -> Path
        """Resolve path to absolute and validate it's within the guard base.
        
        Args:
            path: Path string (relative or absolute)
            
        Returns:
            Resolved Path object
            
        Raises:
            PermissionError: If path escapes the base directory
        """
    
    def is_safe_path(self, path: str) -> bool
        """Check if path is within the guard base.
        
        Args:
            path: Path string
            
        Returns:
            True if path is safe
        """
    
    def list_dir(self, path: str) -> list[str]
        """List directory contents.
        
        Args:
            path: Directory path
            
        Returns:
            List of file/directory names
        """

Usage

guard = DirectoryGuard(Path("~/.local-ai").expanduser())

# Safe: within base
resolved = guard.resolve("sessions/u_s.json")

# Unsafe: escapes base
guard.resolve("../../etc/passwd")  # Raises PermissionError

# Check safety
guard.is_safe_path("sessions/u_s.json")  # True
guard.is_safe_path("../unsafe.txt")  # False

session/store.py - SessionStore

Purpose: Persists session history to JSON files

SessionStore(guard: DirectoryGuard, sessions_dir: Path)

class SessionStore:
    def __init__(self, guard: DirectoryGuard, sessions_dir: Path)
    
    async def append(
        self,
        user_id: str,
        session_id: str,
        message: dict,
    ) -> None
        """Append message to session.
        
        Args:
            user_id: User identifier
            session_id: Session identifier
            message: {"role": "user|assistant", "content": str}
            
        Raises:
            ValueError: If message format invalid
        """
    
    async def get_history(
        self,
        user_id: str,
        session_id: str,
        limit: int = 8,
    ) -> list[dict]
        """Get recent messages from session.
        
        Args:
            user_id: User identifier
            session_id: Session identifier
            limit: Number of messages to return
            
        Returns:
            List of messages from oldest to newest
        """
    
    async def get_session(
        self,
        user_id: str,
        session_id: str,
    ) -> dict | None
        """Get entire session.
        
        Args:
            user_id: User identifier
            session_id: Session identifier
            
        Returns:
            Session dict with "messages" key or None
        """

File Format

{runtime_data_dir}/sessions/{user_id}_{session_id}.json
{
  "messages": [
    {"role": "user", "content": "Hello"},
    {"role": "assistant", "content": "Hi there!"},
    {"role": "user", "content": "How are you?"},
    {"role": "assistant", "content": "I'm doing well..."}
  ]
}

Atomic Writes

Uses .tmp file + rename for atomicity:

# Write to temp file
tmp_path = path.with_suffix(".tmp")
await asyncio.to_thread(tmp_path.write_text, json.dumps(data))

# Atomic rename
await asyncio.to_thread(tmp_path.rename, path)

Concurrent Access

Per-file asyncio.Lock prevents concurrent modification:

self._locks = {}

async def append(self, user_id, session_id, message):
    lock = self._locks.get(f"{user_id}_{session_id}")
    if not lock:
        lock = asyncio.Lock()
        self._locks[f"{user_id}_{session_id}"] = lock
    
    async with lock:
        # ... write logic

packages/local-ai-memory/memory/manager.py — MemoryManager

Purpose: Multi-level runtime memory (L0–L3) under {runtime_data_dir}/memory/. Requires memory.enabled: true.

MemoryManager(guard: DirectoryGuard)

class MemoryManager:
    def __init__(self, guard: DirectoryGuard) -> None

    # L0 — state
    async def get_state(self) -> dict[str, Any]
        """Return state dict. Missing/corrupt → {}."""

    async def update_state(self, patch: dict[str, Any]) -> None
        """Shallow-merge patch into state under per-file lock."""

    # L1 — working memory
    async def get_working_memory(self) -> str
        """Return markdown summary. Missing/corrupt → ""."""

    async def update_working_memory(self, content: str) -> None
        """Atomically overwrite working_memory.md."""

    # L2 — plans
    async def save_plan(self, plan_id: str, payload: dict[str, Any]) -> None
    async def load_plan(self, plan_id: str) -> dict[str, Any]
        """Raises FileNotFoundError if missing or corrupt."""

    # L3 — tool results
    async def save_tool_result(self, result_id: str, payload: dict[str, Any]) -> None
    async def load_tool_result(self, result_id: str) -> dict[str, Any]
        """Raises FileNotFoundError if missing or corrupt."""

RAG knowledge uses {runtime_data_dir}/memory_data/{category}/ (configured via memory.categories).

Storage paths

{runtime_data_dir}/memory/state/{scope}.json
{runtime_data_dir}/memory/working/{scope}.md
{runtime_data_dir}/memory/plans/{user}_{session}.json
{runtime_data_dir}/memory/tool_results/{result_id}.json

Versioned JSON envelope (L0, L2, L3)

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

Plans and tool results use "id", "saved_at", and "payload" instead of "state".

L3 tool result payload (written by ToolExecutor on large successful outputs):

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

The agent receives a compact reference instead of content when UTF-8 size exceeds app.tool_result_inline_limit (default 6144):

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

Summary length is app.tool_result_summary_max_chars (default 1000).

Atomic writes and corruption

Same pattern as SessionStore:

  • Write to {path}.tmp, then Path.replace(target)
  • Corrupt JSON → backup {name}.corrupt.bak, log MEMORY_CORRUPT, safe default
  • Per-path asyncio.Lock for concurrent access
  • Plan/result IDs validated with [A-Za-z0-9_-]+ only

Usage

from memory.manager import MemoryManager

manager = MemoryManager(guard)
await manager.update_state({"current_step": 2})

memory/compressor.py - WorkingMemoryCompressor

Purpose: LLM-summarize conversation state into L1 working_memory.md. Wired in app.py and orchestrated by AgentLoop (not part of MemoryManager storage I/O).

WorkingMemoryCompressor(memory_manager, llm_client, token_counter, target_tokens)

class WorkingMemoryCompressor:
    def __init__(
        self,
        memory_manager: MemoryManager,
        llm_client: LocalLLMClient,
        token_counter: TokenCounter,
        target_tokens: int,
    ) -> None

    @property
    def memory_manager(self) -> MemoryManager

    async def compress(self, messages: list[dict[str, Any]]) -> str
        """Full rebuild from conversation messages. Writes L1; returns markdown.
        On failure, returns existing L1 unchanged."""

    async def update_summary(self, new_messages: list[dict[str, Any]]) -> str
        """Merge new turns with existing L1 via LLM. Enforces target_tokens.
        On failure, returns existing L1 unchanged."""

L1 markdown schema

Required sections (in order): # Current Goal, # Completed, # Important Discoveries, # Open Issues, # Decisions.

Configuration

app key Default
working_memory_target_tokens 768
working_memory_enabled true
working_memory_compact_keep_last 6

Log event

WORKING_MEMORY_COMPRESSED — fields include mode=compress|update, tokens_before, tokens_after.

memory/store.py - MemoryStore

Purpose: Manages Markdown memory files

MemoryStore(guard: DirectoryGuard, memory_data_dir: Path)

class MemoryStore:
    def __init__(self, guard: DirectoryGuard, memory_data_dir: Path)
    
    @property
    def category(self) -> str
        """Memory category (e.g., "knowledge", "personal")"""
    
    @property
    def base_path(self) -> Path
        """Base path for category"""
    
    def list_user_files(self, user_id: str) -> list[str]
        """List all user files in category.
        
        Returns:
            List of filenames with user_id prefix
        """
    
    def get_path(self, user_id: str, filename: str) -> Path
        """Get full path to file.
        
        Args:
            user_id: User identifier
            filename: Filename with user_id prefix
            
        Returns:
            Path object
        """
    
    def read_file(self, path: str) -> str
        """Read file content.
        
        Args:
            path: File path
            
        Returns:
            File content as string
        """
    
    def write_file(self, path: str, content: str) -> str
        """Write file content.
        
        Args:
            path: File path
            content: Content string
            
        Returns:
            Bytes written
        """
    
    def fast_filter(self, query: str, top_n: int = 5) -> list[str]  # alias for bm25_search
    async def bm25_search(self, query: str, top_n: int = 5) -> list[str]
    async def hybrid_search(self, query: str, top_n: int = 5, bm25_weight: float = 0.5) -> list[str]
        """Fast keyword-based filtering.
        
        Args:
            query: Search query
            top_n: Number of results
            
        Returns:
            List of filenames ranked by relevance
        """
    
    def index_exists(self) -> bool
        """Check if index file exists."""
    
    def read_index(self) -> dict | None
        """Read index from disk."""
    
    def write_index(self, index: dict) -> None
        """Write index to disk."""
    
    def populate_index_from_files(
        self,
        user_id: str,
        llm_client: LocalLLMClient,
    ) -> None
        """Build index from all files.
        
        Scans files for title and description,
        uses LLM to generate keywords.
        """

Index Structure

{
  "index_version": 3,
  "last_updated": 1686234567.123,
  "files": [
    {
      "filename": "user1_notes.md",
      "title": "Personal Notes",
      "description": "Collection of personal notes...",
      "keywords": ["notes", "personal", "reminders"]
    }
  ]
}

File Organization

{runtime_data_dir}/memory_data/
├── knowledge/
│   ├── user1_notes.md
│   ├── user2_notes.md
│   └── index.json
├── personal/
│   ├── user1_diary.md
│   ├── user2_diary.md
│   └── index.json
└── work/
    ├── user1_projects.md
    ├── user2_projects.md
    └── index.json

memory/retriever.py - MemoryRetriever

Purpose: 3-stage memory retrieval (BM25/hybrid → LLM select → load)

MemoryRetriever(memory_store: MemoryStore, llm_client: LocalLLMClient, token_counter: TokenCounter)

class MemoryRetriever:
    def __init__(
        self,
        memory_store: MemoryStore,
        llm_client: LocalLLMClient,
        token_counter: TokenCounter,
    )
    
    async def retrieve(
        self,
        query: str,
        user_id: str,
        categories: list[str],
        max_tokens: int = 2000,
    ) -> str
        """Retrieve relevant memory content using 3-stage retrieval.
        
        Stage 1: Fast index filtering (no LLM)
        Stage 2: LLM selects from filtered candidates
        Stage 3: Load selected files within token budget
        
        Args:
            query: User query
            user_id: User identifier
            categories: Memory categories to search
            max_tokens: Maximum tokens to load
            
        Returns:
            Concatenated memory content
            
        Raises:
            TimeoutError: If LLM call times out
        """

Retrieval Flow

# Stage 1: Fast filter
candidates = []
for category in categories:
    store = get_memory_store(category)
    filtered = await store.bm25_search(query, top_n=5)
    candidates.extend(filtered)

# Stage 2: LLM selection
if not candidates:
    return ""

# Send to LLM
prompt = f"""
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: {query}

Top {len(candidates)} candidate files:
- file1.md
- file2.md
...

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

selected_response = await llm.chat([
    {"role": "user", "content": prompt}
])

selected = json.loads(selected_response)
selected = selected[:5]  # Limit to 5

# Stage 3: Load content
chunks = []
used_tokens = 0
for filename in selected:
    content = store.read_file(filename)
    tokens = await token_counter.count(content)
    
    if used_tokens + tokens > max_tokens:
        break
    
    chunks.append(content)
    used_tokens += tokens

return "\n\n".join(chunks)

get_memory_store()

def get_memory_store(category: str) -> MemoryStore:
    """Get MemoryStore for category.
    
    Args:
        category: Category name
        
    Returns:
        MemoryStore instance
    """
    return MemoryStore(guard, memory_data_dir / category)

Error Handling

DirectoryGuard

PermissionError:

guard.resolve("../../etc/passwd")
# Raises PermissionError: Path escapes DirectoryGuard base

SessionStore

ValueError:

await session_store.append(user_id, session_id, {"invalid": "format"})
# Raises ValueError: Invalid message format

MemoryRetriever

TimeoutError:

# LLM call times out
await retriever.retrieve(query, user_id, ["knowledge"])
# Raises TimeoutError

JSONDecodeError:

# LLM returns invalid JSON
selected = json.loads(llm_response)
# Raises JSONDecodeError