3tears-agent-memory

3tears Agent Memory

Memory system for LLM agents. Handles extraction of memorable facts from conversations, hybrid retrieval (semantic + full-text + recency), and memory lifecycle management.

Part of the 3tears framework.

Installation

pip install 3tears-agent-memory

Components

Collections are the single entry point for memory-table SQL

Every memory-table write, single-row read, batch read, and hybrid-search query goes through one of four BaseCollection subclasses; no consumer of this package holds an asyncpg.Pool reference directly.

• MemoriesCollection -- memories table. CRUD through get / save_entity / delete; complex queries through hybrid_search, search_by_ids, search_by_semantic, search_by_fts, find_similar_for_dedup, count_by_user, fetch_content_for_recall.
• MediaCollection -- media parent table. CRUD only.
• MediaContentCollection -- media_content child table. CRUD + hybrid_search, search_by_ids, search_by_semantic, search_by_fts, fetch_content_for_recall.
• MemoryChunkCollection -- memory_chunks child table. CRUD + hybrid_search, search_by_ids, search_by_semantic, fetch_content_for_recall.

All four resolve their L3 pool through CollectionRegistry (same pattern ConversationCollection uses); an L1 SQLiteBackend attached to the registry populates on save_entity and serves subsequent by-id get calls without an L3 round-trip. The media / media_content / memory_chunks tables are introduced by migrations v006 / v007.

Hybrid-search methods carry documented # cache-bypass: <reason> inline comments because the query shape (vector distance, FTS rank, multi-table joins) is not primary-key-addressable and the L1 row cache cannot serve the lookup. Keeping the SQL on the Collection preserves the single entry point. The cache-primitive enforcement walker recognises in-Collection bypass sites as legitimate and reports any bypass that leaks back into retrieval.py / extraction.py / tools.py as a violation.

MemoryExtractor

Extracts memorable facts from conversation turns. Uses a multi-stage pipeline: candidate extraction via LLM, deduplication against existing memories via embedding similarity, and action resolution (ADD / UPDATE / DELETE).

from threetears.agent.memory import (
    MemoriesCollection,
    MemoryConfig,
    MemoryExtractor,
)

extractor = MemoryExtractor(
    config=MemoryConfig(),
    embedding_provider=my_embedding_provider,
    chat_model_factory=my_chat_model_factory,
    authorizer=authorizer_bundle,
    memories_collection=memories_collection,
    summary_callback=on_new_memory,
)

await extractor.extract(
    user_id=user_id,
    conversation_id=conv_id,
    message_id_source=msg_id,
    user_message="I just moved to Portland",
    assistant_response="That's exciting! Portland has great food...",
    turn_count=5,
    agent_id=agent_id,
    customer_id=customer_id,
)

MemoryRetriever

Retrieves relevant memories using hybrid search: pgvector semantic similarity, PostgreSQL full-text search, recency decay, and MMR reranking for diversity. Takes the three search-bearing Collections at construction; no pool.

from threetears.agent.memory import MemoryRetriever, MemoryConfig

retriever = MemoryRetriever(
    config=MemoryConfig(),
    embedding_provider=my_embedding_provider,
    authorizer=authorizer_bundle,
    memories_collection=memories_collection,
    media_content_collection=media_content_collection,
    memory_chunk_collection=memory_chunk_collection,
)

result = await retriever.retrieve_with_candidates(
    user_id,
    "Tell me about Portland",
    agent_id=agent_id,
    customer_id=customer_id,
    caller_user_id=user_id,
    caller_agent_id=agent_id,
)

# result.context     -- formatted string for injection into system prompt
# result.memories    -- raw memory dicts with similarity scores
# result.media_content -- matched media content
# result.memory_chunks -- matched document chunks

Protocols

Implement these to integrate with your infrastructure:

from threetears.agent.memory import EmbeddingProvider, ChatModelFactory

class MyEmbeddingProvider(EmbeddingProvider):
    async def embed(self, text: str) -> tuple[list[float], int, UUID]:
        # Returns (embedding_vector, token_count, model_id)
        ...

class MyChatModelFactory(ChatModelFactory):
    async def create_chat_model(self, purpose: str = "extraction"):
        # Returns a langchain BaseChatModel
        ...

Tools

LangChain tools for agent use: memory search, recall, and explicit add. Factories take Collection references; no pool:

from threetears.agent.memory import (
    load_memory_add_tool,
    load_memory_search_tool,
    load_memory_recall_tool,
)

search_tool = await load_memory_search_tool(
    user_id=user_id,
    embedding_provider=embedding_provider,
    agent_id=agent_id,
    customer_id=customer_id,
    authorizer=authorizer_bundle,
    memories_collection=memories_collection,
    media_content_collection=media_content_collection,
    memory_chunk_collection=memory_chunk_collection,
)
recall_tool = await load_memory_recall_tool(
    user_id=user_id,
    agent_id=agent_id,
    customer_id=customer_id,
    authorizer=authorizer_bundle,
    memories_collection=memories_collection,
    media_content_collection=media_content_collection,
    memory_chunk_collection=memory_chunk_collection,
)
add_tool = await load_memory_add_tool(
    user_id=user_id,
    conversation_id=conv_id,
    message_id=msg_id,
    embedding_provider=embedding_provider,
    agent_id=agent_id,
    customer_id=customer_id,
    authorizer=authorizer_bundle,
    memories_collection=memories_collection,
)

Configuration

from threetears.agent.memory import MemoryConfig

config = MemoryConfig(
    similarity_threshold=0.4,      # minimum cosine similarity for retrieval
    detail_threshold=0.85,         # threshold for including full memory detail
    context_budget=15,             # max memories in context
    dedup_threshold=0.85,          # similarity threshold for deduplication
    max_candidates=10,             # max candidates per extraction
)

Database Schema

Requires PostgreSQL with the pgvector extension. The package's own migration runner (threetears.agent.memory.migrations.register) produces the full schema per agent schema. Registered versions:

• v001 -- memories (PK memory_id, pgvector embedding at the configured width, scoping ids, content, summary, lifecycle timestamps).
• v002 -- conversation_memory_refs (ledger of per-conversation surfaced items).
• v003 -- column reconciliation: renames PK and discriminator to match the package code (id to memory_id, memory_type to type_memory), drops columns the code does not read (embedding_model, importance, metadata, date_accessed), loosens agent_id/customer_id to NULL.
• v004 -- lifecycle + conversation-link columns on memories (conversation_id, message_id_source, is_deleted, media_id, date_deleted, summary) with indexes.
• v005 -- FTS: search_vector TSVECTOR + GIN index + maintenance trigger on memories.
• v006 -- media (parent) + media_content (chunked extracted text with embedding + FTS).
• v007 -- memory_chunks (document-style chunks with heading / page metadata + embedding + FTS).

Every FTS column is trigger-maintained from content + summary (weighted A/B); callers do not have to populate search_vector manually. Integration tests under tests/integration/ exercise the full chain + every public API surface against pgvector/pgvector:pg16 via testcontainers.

Embedding width

memories.embedding, media_content.embedding and memory_chunks.embedding are plain pgvector vector(N) columns, and N is a deployment setting: the width your embedding provider produces. It defaults to 1024 (Voyage AI's voyage-4), so a host that configures nothing is unaffected. Configure it in three places, with the same value:

from threetears.agent.memory.collections import MemoriesCollection, memories_table
from threetears.agent.memory.migrations import register

register(runner, embedding_dim=1536)                       # v001 / v006 / v007 create the columns at 1536
memories = MemoriesCollection(registry, config, authorizer, embedding_dim=1536)
memories_table(metadata, embedding_dim=1536)                # likewise media_content_table, memory_chunks_table

agent-intention takes the same embedding_dim, and should be given the same width, so one provider serves both.

• Ceilings. The column is only useful with its HNSW index. PostgreSQL's pgvector HNSW takes a vector of at most 2,000 dimensions; YugabyteDB's ybhnsw takes up to 16,000 (and has no halfvec operator class, so the column stays vector everywhere). register refuses a width below 1 or above 16,000; the v001 migration refuses a width above 2,000 on PostgreSQL before any statement runs.
• Writes are checked. A save whose embedding is not the collection's width is refused with a ValueError naming both widths, before any cache tier takes the row or any SQL runs.
• An existing schema keeps its width. Migrations run once, so changing embedding_dim later does not change a schema. At start-up, await verify_embedding_width(store, embedding_dim) (from threetears.agent.memory.embedding_width, also exported from the package root) reads each column's declared width from the catalogue and raises EmbeddingWidthMismatch naming the table, the stored width and the configured one.
• Changing the width is an operator action. await resize_embedding_width(store, embedding_dim) drops each column's HNSW index, alters the column to vector(N) with USING NULL, recreates every index under its own name and options, and returns how many embeddings it cleared. Every memory, media content and chunk keeps its content and loses its embedding: re-embed every row with the new model afterwards. Nothing runs it automatically. Run it with writes to these tables stopped: on YugabyteDB the column change is a table rewrite, and writes made while it runs may be lost ("Concurrent DMLs may not be reflected in the new table", yugabyte-db issue 19860). A DataStore runs it on one connection holding the database-wide DDL lock; if it stops part way, running it again recreates any of the package's own indexes that is missing; an index you added yourself is logged, with its full definition, before it is dropped, so it can be recreated by hand.

RBAC Enforcement

Memory reads, writes, and extractions flow through the unified rbac evaluator in threetears.agent.acl. Every (agent, customer) pair is a memory-type namespace in the namespaces table; each access resolves the namespace and evaluates one of three canonical actions against the caller's (user_id, agent_id) pair:

• memory.read -- retrieval / search / recall. Guarded on MemoryRetriever.retrieve*, MemoriesCollection.find_by_user, MemoriesCollection.find_by_scope, the memory_search + memory_recall LangChain tools.
• memory.write -- user-initiated writes. Guarded on MemoriesCollection.save_memory and the memory_add LangChain tool.
• memory.extract -- agent-internal extraction path. Guarded on MemoryExtractor.extract; the owner short-circuit keeps the common case (agent emitting memories on its own namespace) grant-free.

Owner short-circuit: the evaluator allows any action when the calling agent owns the memory namespace. Agent-internal retrieval and extraction therefore work without explicit grants; user-initiated reads and writes require evaluator assignments.

Auto-assignment on first user-write: memory_add ensures a MemoryOwner assignment for the calling user on their first write (idempotent-by-state; the ensurer only fires when the user has zero memory rows in the target schema). Subsequent writes authorize against the materialized grant; admin-revoked grants stay revoked (the ensurer does not resurrect them).

Wiring shape: every consumer of the memory surface REQUIRES a MemoryAuthorizerDependencies bundle exposing:

• acl_cache -- shared threetears.agent.acl.AclCache instance;
• membership_loader + grant_loader -- the evaluator's loaders (threetears.agent.acl.MembershipLoader / GrantLoader);
• namespace_collection -- three-tier NamespaceCollection used to resolve the memory namespace via get_by_owner_and_customer(namespace_type="memory", owner_agent_id, customer_id). This package never writes the row itself;
• namespace_provisioner -- what materializes a missing row. A hub deployment passes HubMemoryNamespaceProvisioner (NATS request/reply; the hub writes the row). A deployment with NO hub, one application owning its own control plane, passes LocalMemoryNamespaceProvisioner(namespace_collection). That writes the same row the hub writes, through NamespaceCollection.ensure_namespace, and trusts the caller's customer, since there is no forwarded identity. None means a missing row is a denial;
• group_collection + group_member_collection + role_collection + role_assignment_collection -- the rbac Collections the first-write owner-assignment path uses via ensure_memory_owner_assignment(...);
• ensure_memory_owner_assignment evicts the rows it writes from acl_cache, so the new grant is honoured on the caller's next request; every other process hears the rows' broadcasts and their tables' write generations.

There is no bypass. Every MemoriesCollection, MemoryRetriever, MemoryExtractor, and LangChain tool factory (load_memory_search_tool, load_memory_add_tool, load_memory_recall_tool) takes the bundle as a required constructor/factory argument; every code path that touches a memory row runs authorize_memory_access first. Callers that omit the bundle fail at the type checker and the Python signature boundary.

• Production wiring builds the bundle directly from the agent-side three-tier stack's Collections (NatsProxyL3Backend-backed NamespaceCollection / GroupCollection / ...).
• Test wiring injects a permissive fixture permissive_memory_authorizer (see tests/conftest.py) that carries in-memory Collection stand-ins and a permissive evaluator. Fixture usage is explicit in every test file that constructs a memory surface.
• Back-office / admin tooling that genuinely needs to read or write memories without an identity must construct its own bundle with Collections bound directly to an asyncpg pool; there is no global escape hatch.

See threetears.agent.memory.authorize for the full public surface.

The three platform roles (MemoryOwner / MemoryReader / MemoryWriter) carry the canonical action vocabulary. Platform-side migrations seed these roles and backfill the rbac rows required for evaluator resolution.