dspy-lancedb-memory 0.1.13

About

DSPy Memory is a persistent vector memory store for DSPy-powered AI agents. It uses DSPy signatures to extract structured, categorized memories from conversation turns and stores them in LanceDB for efficient semantic retrieval.

• Method-based SDK — Single memory.configure() entry point for extraction LM, embedding model, and reranker
• DSPy-native extraction — Uses ChainOfThought with a typed ExtractMemory signature to pull salient information from conversations
• Structured memory taxonomy — Six memory categories (preference, semantic, episodic, procedural, summary, artifact) for fine-grained organization
• Persistent vector storage — LanceDB-backed with automatic text embeddings via the DSPy Embedder
• Semantic search — Query memories by user ID, session ID, conversation ID, memory type, or natural language
• Hybrid search — Combine vector similarity with full-text search for better recall on keyword-heavy queries
• Custom scope and metadata — Attach arbitrary ownership dimensions and structured filters to every memory
• Bound stores — store.with_scope() returns a BoundMemoryStore with user/session/scope pre-bound
• Optional reranking — LiteLLMReranker wraps litellm.rerank() for cross-encoder reranking via Cohere, Jina, and any LiteLLM-compatible provider
• Full CRUD — Create, search, update, and delete individual memories or batch-extract from conversations

(back to top)

Quick Start

Prerequisites

Set the API key environment variable for your chosen provider. LiteLLM routes to the correct key automatically based on the model prefix:

# Examples — set whichever matches your provider
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export CO_API_KEY="..."            # Cohere reranker
export JINA_API_KEY="..."          # Jina reranker

If you use a proxy or gateway (e.g. OpenRouter, LiteLLM proxy), set the base URL and key:

export OPENAI_API_BASE="https://openrouter.ai/api/v1"
export OPENAI_API_KEY="sk-or-v1-..."

Install

Install dspy-lancedb-memory with uv (recommended)

uv add dspy-lancedb-memory

Install with pip (alternative)

pip install dspy-lancedb-memory

(back to top)

Usage

Basic Usage

from dspy_lancedb_memory import memory
import dspy

# Configure: extraction LM, embedding LM, and optional reranker LM
memory.configure(
    extraction_lm=dspy.LM("openai/gpt-4o-mini"),
    embedding_lm=dspy.LM("openai/text-embedding-3-small"),
    reranker_lm=dspy.LM("cohere/rerank-4-fast"),   # optional — omit to disable
)

# Create a store — picks up all defaults from configure()
store = memory.Store()

# Store a single memory
store.create_memory(
    user_id="user_123",
    content="Edward prefers DSPy signatures over ad-hoc prompts.",
    memory_type="preference",
)

# Store with session and conversation scoping
store.create_memory(
    user_id="user_123",
    session_id="session_abc",
    conversation_id="conv_456",
    content="Remember this from our conversation about RAG pipelines.",
    memory_type="episodic",
)

# Search memories (with reranking)
results = store.search_memories(
    user_id="user_123",
    session_id="session_abc",   # optional — narrow to a session
    query="What does Edward prefer?",
    use_reranker=True,
)
print(results)

Extract Memories from Conversation

The real power is automatic extraction. Pass a conversation turn and the LLM extracts all salient memories, each categorized by type.

from dspy_lancedb_memory import memory
import dspy

memory.configure(extraction_lm=dspy.LM("openai/gpt-4o-mini"))
store = memory.Store()

messages = [
    {
        "role": "user",
        "content": "I really like using DSPy signatures instead of writing prompts by hand. "
                   "I'm building a CLI tool to organize my digital bookmarks by topic. "
                   "The PR is at github.com/example/bookmark-organizer/pull/42.",
    },
]

created = store.create_memories(
    user_id="user_123",
    contents=messages,
    extract=True,  # default; uses DSPy ChainOfThought to extract memories
)
# Returns multiple MemoryItems categorized automatically:
#   preference: "User prefers DSPy signatures over writing prompts by hand"
#   semantic: "User is building a CLI bookmark organizer"
#   procedural: "User is organizing bookmarks by topic"
#   artifact: "github.com/example/bookmark-organizer/pull/42"

for m in created:
    print(f"[{m['memory_type']}] {m['content']}")

Search with Memory Type and Reranking

from dspy_lancedb_memory import memory
import dspy

memory.configure(
    extraction_lm=dspy.LM("openai/gpt-4o-mini"),
    reranker_lm=dspy.LM("cohere/rerank-4-fast"),
)
store = memory.Store()

# Filter by memory type and optionally use reranking for better results
results = store.search_memories(
    user_id="user_123",
    query="What are Edward's tool preferences?",
    memory_type="preference",  # optional filter
    limit=5,
    use_reranker=True,         # uses configured reranker endpoint
)

Filtering by Session and Conversation

Every memory can be scoped to a session_id and conversation_id. Both are optional and can be used independently or together.

# Scope to a specific session
session_memories = store.search_memories(
    user_id="user_123",
    session_id="session_abc",
    query="What did we discuss last time?",
)

# Scope to a specific conversation
conversation_memories = store.search_memories(
    user_id="user_123",
    conversation_id="conv_456",
    query="RAG pipeline details",
)

# Combine session + conversation for maximum precision
precise = store.search_memories(
    user_id="user_123",
    session_id="session_abc",
    conversation_id="conv_456",
    query="specific topic",
)

# Omit both to search across all sessions and conversations
all_results = store.search_memories(
    user_id="user_123",
    query="anything",
)

Custom Scope and Metadata Filters

Use scope for custom ownership or routing dimensions beyond user_id, session_id, and conversation_id. Scope is returned as memory.scope and can be used during search, upsert, delete-by-search, and process flows.

store.create_memory(
    user_id="user_123",
    content="Project uses LanceDB for vector memory.",
    memory_type="project_note",
    scope={
        "tenant": "acme",
        "repo": "rag-agent",
        "environment": "prod",
    },
    metadata={"source": "chat", "priority": "high"},
)

results = store.search_memories(
    user_id="user_123",
    query="vector memory setup",
    scope={"repo": "rag-agent", "environment": "prod"},
    metadata_filter={"source": "chat"},
)

Custom extraction signatures may also return per-memory metadata on each MemoryItem. That metadata is merged with the call-level metadata passed to create_memories or upsert_memories.

Use Scope when you want a typed helper instead of a raw dict:

from dspy_lancedb_memory import Scope

scope = Scope(tenant="acme", repo="rag-agent", environment="prod")
store.create_memory(
    user_id="user_123",
    content="Production repo uses LanceDB for memory.",
    scope=scope,
)

Bind repeated context once with with_scope():

project_store = store.with_scope(
    user_id="user_123",
    scope={"tenant": "acme", "repo": "rag-agent"},
)

project_store.create_memory(content="Project uses LanceDB.")
project_store.search_memories(query="vector store")

Metadata and scope filters support equality by default plus the full set of comparison operators:

Multiple operators on the same field are AND-ed together:

results = store.search_memories(
    user_id="user_123",
    query="important project notes",
    metadata_filter={
        "priority": {"in": ["high", "urgent"]},
        "confidence": {"gte": 0.8},
        "tags": {"contains": "backend"},
        "reviewed": {"exists": True},
    },
)

Hybrid Search

Combine vector similarity with full-text search for better recall on keyword-heavy queries. A full-text search index on the content column is created automatically.

results = store.search_memories(
    user_id="user_123",
    query="RAG pipeline configuration",
    query_type="hybrid",   # combines vector + full-text search
)

Use refine_factor to re-score hybrid candidates with a second vector pass — higher values improve ranking quality at the cost of latency:

results = store.search_memories(
    user_id="user_123",
    query="RAG pipeline configuration",
    query_type="hybrid",
    refine_factor=10,      # re-rank top candidates with refined vectors
    use_reranker=True,     # chain with cross-encoder reranking
)

refine_factor also works with the default query_type="vector" to refine pure vector results.

Raw Store (No Extraction)

Store content verbatim without LLM extraction.

store.create_memories(
    user_id="user_123",
    contents=[{"role": "user", "content": "A raw fact worth storing."}],
    extract=False,
    memory_type="semantic",
)

Update and Delete

# Update memory content (re-embeds automatically)
store.update_memory(
    memory_id="some-uuid",
    content="Updated memory text",
)

# Patch metadata or scope while preserving append-only history
store.update_memory_metadata(
    memory_id="some-uuid",
    metadata={"priority": "high"},
)

store.update_memory_scope(
    memory_id="some-uuid",
    scope={"project_id": "new-project"},
)

# Deterministic inspection without vector search
memory = store.get_memory(memory_id="some-uuid")
all_project_memories = store.list_memories(
    user_id="user_123",
    scope={"project_id": "rag-agent"},
)
history = store.get_memory_history(memory_id="some-uuid")

# Delete a memory
store.delete_memory(memory_id="some-uuid")

Upsert — Insert, Update, or Skip

upsert_memory uses semantic similarity to decide what to do:

1. Exact match — same content string exists → skip (no-op)
2. Semantic match — similar content found (cosine similarity ≥ threshold) → update it
3. No match — nothing close enough → insert a new memory

# First insert
store.upsert_memory(
    user_id="user_123",
    content="Edward is building a RAG pipeline for climate modeling.",
    memory_type="semantic",
)

# Same content string → skip (returns existing row unchanged)
store.upsert_memory(
    user_id="user_123",
    content="Edward is building a RAG pipeline for climate modeling.",
)

# Semantically similar content → update that memory in place
store.upsert_memory(
    user_id="user_123",
    content="Edward is designing a RAG pipeline for climate data analysis.",
    similarity_threshold=0.8,  # lower = more aggressive updates
)

# Completely different content → insert a new row
store.upsert_memory(
    user_id="user_123",
    content="Edward prefers DSPy signatures over raw prompts.",
    memory_type="preference",
)

The similarity_threshold (default 0.85) controls how close two memories must be to consider them the same. Higher values make upsert more conservative (mostly inserts); lower values make it more aggressive (mostly updates).

Batch Upsert with Extraction

upsert_memories mirrors create_memories exactly — same parameters, same DSPy extraction — but each extracted memory goes through the upsert decision instead of a blind insert.

from dspy_lancedb_memory import memory
import dspy

memory.configure(extraction_lm=dspy.LM("openai/gpt-4o-mini"))
store = memory.Store()

messages = [
    {
        "role": "user",
        "content": "I really like using DSPy signatures instead of writing prompts by hand. "
                   "I'm working on a RAG pipeline for my thesis on climate modeling. "
                   "The PR is at github.com/example/climate-rag/pull/42.",
    },
]

upserted = store.upsert_memories(
    user_id="user_123",
    contents=messages,
    extract=True,
)

for m in upserted:
    # Each extracted memory was independently upserted:
    #   - exact match → skip
    #   - semantic match → update
    #   - no match → insert
    print(f"[{m['memory_type']}] {m['content']}")

Using the Reranker

The easiest way — configure via memory.configure(reranker_lm=...) with a dspy.LM and memory.Store() picks it up automatically:

from dspy_lancedb_memory import memory
import dspy

memory.configure(
    extraction_lm=dspy.LM("openai/gpt-4o-mini"),
    reranker_lm=dspy.LM("cohere/rerank-4-fast"),
)
store = memory.Store()  # LiteLLMReranker auto-created from reranker_lm

You can also pass a plain model string instead of a dspy.LM:

memory.configure(reranker_lm="cohere/rerank-english-v3.0")

For full control (custom column, top_n, etc.), build a LiteLLMReranker and pass it to Store():

from dspy_lancedb_memory import LiteLLMReranker

reranker = LiteLLMReranker(
    model="cohere/rerank-english-v3.0",
    column="content",               # LanceDB column to rerank against
    top_n=20,                       # optional: limit reranked candidates
)

store = memory.Store(reranker=reranker)

The model string uses the same provider/model format as dspy.LM — e.g. "cohere/rerank-english-v3.0", "jina/jina-reranker-v2-base-multilingual". LiteLLM handles the routing.

Custom API Base and Key

When running behind a proxy, gateway, or self-hosted endpoint, pass api_base and api_key directly to LiteLLMReranker:

from dspy_lancedb_memory import LiteLLMReranker

reranker = LiteLLMReranker(
    model="my-provider/rerank-model",
    api_base="https://my-gateway.example.com/v1",
    api_key="sk-my-secret",
    column="content",
    top_n=20,
)

store = memory.Store(reranker=reranker)

For embeddings behind a custom endpoint, pass the same api_base/api_key via a dspy.LM:

import dspy

memory.configure(
    embedding_lm=dspy.LM(
        "openai/text-embedding-3-small",
        api_base="https://my-gateway.example.com/v1",
        api_key="sk-my-secret",
    ),
)

LiteLLM automatically routes to the correct provider based on the model prefix. If your provider is not in LiteLLM's built-in list, LiteLLMReranker falls back to calling a Cohere-compatible /rerank endpoint on your api_base.

Custom Configuration

Everything — including LanceDB defaults — in one call:

from dspy_lancedb_memory import memory
import dspy

memory.configure(
    extraction_lm=dspy.LM("anthropic/claude-sonnet-4-20250514"),           # extraction LM
    embedding_lm=dspy.LM("openai/text-embedding-3-small"),              # embedding LM
    embedding_dim=1536,                                                             # must match
    reranker_lm=dspy.LM("cohere/rerank-4-fast"),                         # reranker model
    uri=".my_memories",                                                              # LanceDB path
    table_name="user_memories",                                                      # LanceDB table
)

store = memory.Store()  # everything inherited from configure()

Override individual fields on Store() when you need something different:

store = memory.Store(uri="./scratch", reranker=None)

(back to top)

Memory Taxonomy

Every extracted memory is categorized into one of six types:

When storing directly (without extraction), the default type is semantic.

(back to top)

Available API

(back to top)

Development

Code Quality

This project uses several tools to maintain code quality:

• Ruff: Linting and formatting
• isort: Import sorting
• pytest: Testing framework
• deptry: Dependency checking
• ty: Type checking (based on pyright)

Available commands:

# Run all quality checks
uv run poe clean-full

# Individual checks
uv run poe lint          # Ruff linting
uv run poe format        # Ruff formatting
uv run poe sort          # Import sorting

Testing

Run tests using pytest:

# Run all tests
uv run pytest

# Run specific test
uv run pytest path/to/test.py::test_name

(back to top)

Contributing

Quick workflow:

1. Fork and branch: git checkout -b feature/name
2. Make changes
3. Run checks: uv run poe clean-full
4. Commit and push
5. Open a Pull Request

(back to top)

License

MIT (as declared in pyproject.toml).

(back to top)