haive.core.tools.store_manager

Store management system for Haive agents.

This module provides a comprehensive store management system similar to LangMem, with tools for storing, retrieving, and managing agent memories using our PostgreSQL store infrastructure.

Classes

MemoryEntry

A single memory entry in the store.

StoreManager

Centralized store management for agent memories.

Module Contents

class haive.core.tools.store_manager.MemoryEntry(/, **data)[source]

Bases: pydantic.BaseModel

A single memory entry in the store.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

classmethod from_store_value(value)[source]

Create from store dictionary.

Parameters:

value (dict[str, Any])

Return type:

MemoryEntry

to_store_value()[source]

Convert to store-compatible dictionary.

Return type:

dict[str, Any]

class haive.core.tools.store_manager.StoreManager(store=None, default_namespace=None, store_config=None)[source]

Centralized store management for agent memories.

This class provides a high-level interface for managing agent memories using the Haive store infrastructure, similar to LangMem but customized for our architecture.

Features: - Namespace-based memory isolation (user_id, agent_id, session_id) - Semantic search capabilities - Memory categorization and tagging - Importance-based retrieval - Automatic metadata management

Initialize the store manager.

Parameters:
  • store (haive.core.persistence.store.base.SerializableStoreWrapper | None) – Pre-configured store wrapper (optional)

  • default_namespace (tuple[str, Ellipsis] | None) – Default namespace for operations

  • store_config (dict[str, Any] | None) – Configuration for creating a new store

create_agent_namespace(agent_id, user_id=None)[source]

Create an agent-specific namespace.

Parameters:
  • agent_id (str) – Agent identifier

  • user_id (str | None) – Optional user identifier for user-agent isolation

Returns:

Namespace tuple for the agent

Return type:

tuple[str, Ellipsis]

create_session_namespace(session_id, agent_id=None, user_id=None)[source]

Create a session-specific namespace.

Parameters:
  • session_id (str) – Session identifier

  • agent_id (str | None) – Optional agent identifier

  • user_id (str | None) – Optional user identifier

Returns:

Namespace tuple for the session

Return type:

tuple[str, Ellipsis]

create_user_namespace(user_id)[source]

Create a user-specific namespace.

Parameters:

user_id (str) – User identifier

Returns:

Namespace tuple for the user

Return type:

tuple[str, Ellipsis]

delete_memory(memory_id, namespace=None)[source]

Delete a memory.

Parameters:
  • memory_id (str) – Memory ID to delete

  • namespace (tuple[str, Ellipsis] | None) – Storage namespace

Returns:

True if deleted, False if not found

Return type:

bool

get_memory_stats(namespace=None)[source]

Get statistics about stored memories.

Parameters:

namespace (tuple[str, Ellipsis] | None) – Storage namespace

Returns:

Dictionary with memory statistics

Return type:

dict[str, Any]

list_memories_by_category(category, namespace=None, limit=50)[source]

List memories by category.

Parameters:
  • category (str) – Category to filter by

  • namespace (tuple[str, Ellipsis] | None) – Storage namespace

  • limit (int) – Maximum results

Returns:

List of memories in the category

Return type:

list[MemoryEntry]

retrieve_memory(memory_id, namespace=None)[source]

Retrieve a specific memory by ID.

Parameters:
  • memory_id (str) – The memory ID to retrieve

  • namespace (tuple[str, Ellipsis] | None) – Storage namespace

Returns:

MemoryEntry if found, None otherwise

Return type:

MemoryEntry | None

search_memories(query, category=None, min_importance=None, tags=None, limit=10, namespace=None)[source]

Search memories using semantic search.

Parameters:
  • query (str) – Search query

  • category (str | None) – Filter by category

  • min_importance (float | None) – Minimum importance score

  • tags (list[str] | None) – Required tags

  • limit (int) – Maximum results

  • namespace (tuple[str, Ellipsis] | None) – Storage namespace

Returns:

List of matching memories

Return type:

list[MemoryEntry]

store_memory(content, category='general', importance=0.5, tags=None, metadata=None, namespace=None, memory_id=None)[source]

Store a new memory.

Parameters:
  • content (str) – The memory content

  • category (str) – Memory category (e.g., “user_preference”, “fact”, “event”)

  • importance (float) – Importance score 0-1

  • tags (list[str] | None) – Optional tags for the memory

  • metadata (dict[str, Any] | None) – Additional metadata

  • namespace (tuple[str, Ellipsis] | None) – Storage namespace (defaults to default_namespace)

  • memory_id (str | None) – Optional custom memory ID

Returns:

Memory ID

Return type:

str

update_memory(memory_id, content=None, category=None, importance=None, tags=None, metadata=None, namespace=None)[source]

Update an existing memory.

Parameters:
  • memory_id (str) – Memory ID to update

  • content (str | None) – New content (optional)

  • category (str | None) – New category (optional)

  • importance (float | None) – New importance score (optional)

  • tags (list[str] | None) – New tags (optional)

  • metadata (dict[str, Any] | None) – New metadata (optional)

  • namespace (tuple[str, Ellipsis] | None) – Storage namespace

Returns:

True if updated, False if memory not found

Return type:

bool