MemoryLens

Core SDK — memorylens._core

The core layer is the heart of MemoryLens. It defines the trace schema, manages the tracer lifecycle, implements context propagation, and runs the span processor pipeline. Everything else in the system builds on top of these primitives.

Trace Schema and Data Model

MemorySpan

MemorySpan is a frozen, slotted dataclass — immutable once finalized. Every memory operation produces exactly one MemorySpan. The schema is the single source of truth for all downstream consumers: exporters, the CLI, the web UI, and the analysis tools.

# src/memorylens/_core/span.py
@dataclass(frozen=True, slots=True)
class MemorySpan:
    """A single traced memory operation."""

    # Identity
    span_id: str                # unique hex UUID
    trace_id: str               # groups spans in one logical operation
    parent_span_id: str | None  # for nested operations

    # Classification
    operation: MemoryOperation  # WRITE, READ, COMPRESS, UPDATE
    status: SpanStatus          # OK, ERROR, DROPPED

    # Timing (epoch nanoseconds)
    start_time: float
    end_time: float
    duration_ms: float          # computed: (end - start) / 1_000_000

    # Context (inherited from MemoryContext)
    agent_id: str | None
    session_id: str | None
    user_id: str | None

    # Memory content (redactable via MEMORYLENS_CAPTURE_CONTENT=false)
    input_content: str | None
    output_content: str | None

    # Operation-specific attributes (free-form dict)
    attributes: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        """Serialize to plain dict. Enum values become their string values."""

MemoryOperation Enum

The four fundamental operations of any memory system, modeled as a str enum so values serialize naturally:

SpanStatus Enum

Operation-Specific Attributes

The attributes dict carries operation-type-specific fields. These map 1:1 to OTLP span attributes with a memorylens. prefix:

TracerProvider and Tracer

The TracerProvider is a singleton that holds global configuration: the processor pipeline, the sampler, and the service name. All instrumentation flows through it.

# src/memorylens/_core/tracer.py
class TracerProvider:
    """Singleton that manages tracers, processors, and sampling."""
    _instance: TracerProvider | None = None

    def __init__(self) -> None:
        self.processors: list[SpanProcessor] = []
        self.sampler = Sampler(rate=1.0)
        self.service_name: str = "memorylens"

    def add_processor(self, processor: SpanProcessor) -> None: ...
    def get_tracer(self, name: str) -> Tracer: ...
    def shutdown(self) -> None: ...

    @classmethod
    def get(cls) -> TracerProvider: ...  # singleton accessor

    @classmethod
    def reset(cls) -> None: ...  # for testing — shuts down and clears instance

The Tracer is created per-module (by name) via provider.get_tracer(name). It provides a context manager for creating spans:

class Tracer:
    @contextmanager
    def start_span(
        self,
        operation: MemoryOperation,
        parent_span_id: str | None = None,
        attributes: dict[str, Any] | None = None,
    ) -> Generator[_MutableSpan, None, None]:
        # 1. Check sampler — if not sampled, yield a no-op span
        # 2. Read current MemoryContext for agent/session/user IDs
        # 3. Create _MutableSpan with new trace_id and span_id (UUID hex)
        # 4. Call on_start() on all processors
        # 5. yield span — caller can set attributes, content, status
        # 6. On exception: set status=ERROR, capture error.type/error.message
        # 7. On exit: finalize() → frozen MemorySpan, call on_end() on all processors

The internal _MutableSpan builder accumulates state during the span lifecycle. It exposes set_attribute(), set_status(), and set_content() methods. finalize() stamps end_time, computes duration_ms, and returns an immutable MemorySpan.

Decorators

The four instrument_* decorators are the primary instrumentation API. They wrap existing functions without requiring any modification to the function body.

# src/memorylens/_core/decorators.py

@instrument_write(backend="mem0", capture_content=True)
def store_memory(user_id: str, content: str) -> bool: ...

@instrument_read(backend="mem0", top_k=5)
def search_memories(query: str) -> list[Memory]: ...

@instrument_compress(model="gpt-4o-mini")
def summarize(memories: list[str]) -> str: ...

@instrument_update(backend="mem0")
def update_memory(memory_id: str, new_content: str) -> bool: ...

All four decorators are implemented via a shared _make_decorator(operation, **kwargs) factory. The wrapper function:

1. Gets the TracerProvider singleton and acquires a tracer for the wrapped function's module
2. Opens a span with the correct MemoryOperation and any static keyword arguments as initial attributes
3. Conditionally captures repr(args/kwargs) as input_content (if capture_content=True)
4. Calls the original function
5. Conditionally captures repr(result) as output_content
6. Returns the result — exceptions propagate naturally after setting status=ERROR

Content capture resolution order: explicit decorator kwarg → MEMORYLENS_CAPTURE_CONTENT env var → default True.

Context Propagation

The MemoryContext is a frozen dataclass that acts as a Python context manager. It uses ContextVar (from the standard library contextvars module) to propagate agent/session/user metadata through the call stack without modifying function signatures. It is fully async-safe because ContextVar isolates state per asyncio Task.

# src/memorylens/_core/context.py
@dataclass(frozen=True, slots=True)
class MemoryContext:
    agent_id: str | None = None
    session_id: str | None = None
    user_id: str | None = None

    def __enter__(self) -> Self:
        # sets the ContextVar token
        object.__setattr__(self, "_token", _current_context.set(self))
        return self

    def __exit__(self, *exc) -> None:
        # resets to the previous context (safe nesting)
        _current_context.reset(self._token)

# Usage:
with memorylens.context(agent_id="support-bot", session_id="sess-123"):
    store_memory("user prefers vegetarian meals")
    # → span will have agent_id="support-bot", session_id="sess-123"

The public memorylens.context() function wraps MemoryContext constructor. The Tracer.start_span() calls get_current_context() to read the active context at span creation time.

SpanProcessor Pipeline

Span processors sit between the tracer and the exporters. They receive spans via on_start() (before execution) and on_end() (after execution). Multiple processors can be registered; they are called in registration order.

class SpanProcessor(Protocol):
    def on_start(self, span: MemorySpan) -> None: ...
    def on_end(self, span: MemorySpan) -> None: ...
    def shutdown(self) -> None: ...
    def force_flush(self, timeout_ms: int = 30000) -> bool: ...

SimpleSpanProcessor

Synchronous: calls exporter.export([span]) immediately on on_end(). Used for debugging, testing, and scenarios where latency is not a concern.

BatchSpanProcessor

The production processor. Appends spans to a bounded deque (max 2048) under a lock. A daemon thread wakes every 5 seconds (or when the queue reaches 512 spans, or on force_flush()) and exports in batches. This keeps the hot path non-blocking.

class BatchSpanProcessor:
    def __init__(
        self,
        exporter: SpanExporter,
        max_batch_size: int = 512,       # trigger flush when queue hits this
        schedule_delay_ms: int = 5000,   # background flush interval
        max_queue_size: int = 2048,      # drops oldest spans if exceeded
    ) -> None: ...

On shutdown(), the processor signals the background thread, waits up to 10 seconds, flushes any remaining spans, then shuts down the exporter.

Sampler

The Sampler is a simple rate-based sampler. It generates a random float on each call to should_sample() and returns True if it is below the configured rate. Rate is set via memorylens.init(sample_rate=0.1) or MEMORYLENS_SAMPLE_RATE env var. Default is 1.0 (sample everything). When a span is not sampled, the tracer yields a no-op _MutableSpan with empty IDs and skips all processor calls.

Exporters — memorylens._exporters

Exporters are the output stage of the pipeline. They receive batches of MemorySpan objects from the processor and persist them to a backend. Three exporters ship with MemoryLens core.

SpanExporter Protocol

class SpanExporter(Protocol):
    def export(self, spans: list[MemorySpan]) -> ExportResult: ...
    def shutdown(self) -> None: ...

class ExportResult(Enum):
    SUCCESS = "success"
    FAILURE = "failure"

SQLite Exporter

The primary local storage backend. Writes spans to a SQLite database at ~/.memorylens/traces.db (configurable). Designed for developer use: zero server setup, persistent across sessions, queryable with standard SQL.

Database Schema

-- spans table (created on first use)
CREATE TABLE IF NOT EXISTS spans (
    span_id        TEXT PRIMARY KEY,
    trace_id       TEXT NOT NULL,
    parent_span_id TEXT,
    operation      TEXT NOT NULL,        -- "memory.write" etc.
    status         TEXT NOT NULL,        -- "ok" / "error" / "dropped"
    start_time     REAL NOT NULL,        -- epoch nanoseconds
    end_time       REAL NOT NULL,
    duration_ms    REAL NOT NULL,
    agent_id       TEXT,
    session_id     TEXT,
    user_id        TEXT,
    input_content  TEXT,
    output_content TEXT,
    attributes     TEXT NOT NULL DEFAULT '{}'  -- JSON blob
)

-- Indexes for common query patterns
CREATE INDEX IF NOT EXISTS idx_spans_trace_id   ON spans (trace_id)
CREATE INDEX IF NOT EXISTS idx_spans_session_id ON spans (session_id)
CREATE INDEX IF NOT EXISTS idx_spans_operation  ON spans (operation)
CREATE INDEX IF NOT EXISTS idx_spans_start_time ON spans (start_time)

The connection is opened in WAL mode (PRAGMA journal_mode=WAL) for concurrent read safety. check_same_thread=False is set because the BatchSpanProcessor writes from a background thread.

Query Methods

Compression Audits Table

Created lazily on the first save_audit() call (not at initialization):

CREATE TABLE IF NOT EXISTS compression_audits (
    span_id             TEXT PRIMARY KEY,
    semantic_loss_score REAL NOT NULL,
    compression_ratio   REAL NOT NULL,
    pre_sentence_count  INTEGER NOT NULL,
    post_sentence_count INTEGER NOT NULL,
    sentences           TEXT NOT NULL,    -- JSON array of SentenceAnalysis dicts
    scorer_backend      TEXT NOT NULL,    -- "local" or "openai"
    created_at          REAL NOT NULL     -- epoch timestamp
)

OTLP Exporter

Translates MemorySpan objects to OpenTelemetry spans using a _ReadableSpanAdapter that presents the MemoryLens schema as an OTel-compatible interface. This allows using the upstream OTLPSpanExporter from the OpenTelemetry SDK directly.

_ReadableSpanAdapter

The adapter presents MemorySpan fields as OTel span properties:

The OTLP exporter is configured via standard OTel environment variables: OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_EXPORTER_OTLP_HEADERS, and OTEL_EXPORTER_OTLP_PROTOCOL. The default endpoint is http://localhost:4317.

JSONL Exporter

The simplest exporter — writes one JSON object per line to stdout or a file path. Stateless. Used by memorylens traces export and as a portable format for piping to other tools. Each line is the output of span.to_dict().

Exporter Registry and Factory

# src/memorylens/_exporters/__init__.py
def create_exporter(name: str, **kwargs) -> SpanExporter:
    match name:
        case "sqlite": return SQLiteExporter(**kwargs)
        case "otlp":   return OTLPExporter(**kwargs)
        case "jsonl":  return JSONLExporter(**kwargs)
        case _: raise ValueError(f"Unknown exporter: {name}")

Framework Integrations — memorylens.integrations

Auto-instrumentation patches framework memory classes at runtime so that zero code changes are required in user applications. Each integration follows an identical Instrumentor protocol.

Instrumentor Protocol

class Instrumentor(Protocol):
    def instrument(self, **kwargs) -> None: ...
    def uninstrument(self) -> None: ...

Each instrumentor stores the original method references so uninstrument() can restore the original behavior. The integration registry maps names to instrumentor classes:

# memorylens.init(instrument=["langchain", "mem0"]) resolves via:
def create_instrumentor(name: str) -> Instrumentor:
    return _REGISTRY[name]()  # raises helpful error if framework not installed

Auto-Instrumentation Shorthand

# Initialize all integrations in one call:
memorylens.init(instrument=["langchain", "mem0", "llamaindex"])

# Or initialize individual instrumentors explicitly:
from memorylens.integrations.langchain import LangChainInstrumentor
LangChainInstrumentor().instrument()

# To stop instrumentation:
from memorylens.integrations.mem0 import Mem0Instrumentor
inst = Mem0Instrumentor()
inst.instrument()
# ... later ...
inst.uninstrument()

If a framework is not installed, a clear error is raised: "LangChain not found. Install with: pip install memorylens[langchain]".

Web Dashboard — memorylens._ui

The MemoryLens web dashboard is a local developer tool that provides a browser-based interface for trace inspection. It launches with a single CLI command and reads from the same SQLite database that the SDK writes to. It is available as an optional extra: pip install memorylens[ui].

Technology Stack

• FastAPI — async Python web framework; app factory pattern for testability
• Jinja2 — server-side HTML templating; templates live in _ui/templates/
• htmx — loaded via CDN; enables AJAX interactions without JavaScript code
• TailwindCSS — loaded via CDN; utility-first styling, dark theme support
• uvicorn — ASGI server; binds to 127.0.0.1 only (local-only access)

App Factory

# src/memorylens/_ui/server.py
def create_app(db_path: str = _DEFAULT_DB, ingest: bool = False) -> FastAPI:
    app = FastAPI(title="MemoryLens", docs_url=None, redoc_url=None)
    # Mount static files at /static
    # Configure Jinja2 templates
    # Create SQLiteExporter(db_path) — shared via app.state
    # Register: create_trace_routes(app)
    # Register: create_compression_routes(app)
    if ingest:
        # Register: create_ingest_routes(app)  — OTLP HTTP receiver
    return app

def run(db_path, port=8000, ingest=False) -> None:
    app = create_app(db_path, ingest)
    uvicorn.run(app, host="127.0.0.1", port=port, log_level="warning")

Content Negotiation

All filterable endpoints serve both full HTML pages and htmx partials from the same URL. htmx automatically sends an HX-Request: true header on all requests it triggers. The routes check for this header (or use separate /api/ paths for partials) and return the appropriate template.

API Endpoints

Trace List View (/traces)

The main landing page. A filterable, paginated table of all memory operation traces.

• Filter bar: search input (debounced 300ms), operation dropdown, status dropdown, agent ID input
• Table columns: Trace ID, Operation (color-coded badge), Status (dot indicator), Duration, Agent, Session, Content Preview, Time (relative)
• Error rows receive a subtle red background tint
• Pagination: "Showing 1–50 of 247" with Prev/Next using htmx offset parameter
• htmx live tail: when --ingest is active, hx-trigger="every 2s" polls for new traces

Trace Detail View (/traces/{trace_id})

Deep-dive into a single span. Two-column layout:

• Left column: span timeline bar (colored by operation type), input/output content blocks (monospace), error block (red-tinted, only for error/dropped spans)
• Right column: attributes panel split into standard fields and custom attributes; action buttons including "Export JSON" and context-specific debug links ("Debug Retrieval" for READ spans, "Debug Compression" for COMPRESS spans)

Retrieval Debugger (/traces/{trace_id}/retrieval)

The flagship debugging view for READ operations. Visualizes exactly why a retrieval returned or missed specific memories:

• Query section: search query text + parameters panel (backend, top_k, threshold, results_count)
• Score visualization: horizontal bars for every candidate memory scaled by similarity score (0.0–1.0). Green bars with "RETURNED" badge above threshold; red bars with "FILTERED" badge below. A dashed yellow vertical threshold line runs through all bars
• Near-miss callout: amber box that appears when candidates scored within 0.10 of the threshold, with suggested actions
• Data source: all data comes from existing READ span attributes (scores, threshold, top_k, results_count) — no new SDK changes required

OTLP Ingest Endpoint (POST /v1/traces)

When started with --ingest, the dashboard doubles as a lightweight OTLP HTTP/JSON collector. This allows agent applications to point their OTEL_EXPORTER_OTLP_ENDPOINT directly at the dashboard:

# Agent configuration for live ingest:
export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:8000
export OTEL_EXPORTER_OTLP_PROTOCOL=http/json

The ingest handler parses ExportTraceServiceRequest JSON, iterates resourceSpans[].scopeSpans[].spans[], skips spans without a memorylens.operation attribute, maps OTel attributes back to MemorySpan fields, and writes to SQLite. The UI then reflects new traces within the live poll interval.

HTTP/JSON only (no gRPC) — gRPC would require protobuf compilation on the server side, adding unnecessary complexity for a local dev tool.

File Structure

Compression Auditor — memorylens._audit

The Compression Auditor performs offline semantic analysis of COMPRESS spans, determining what information was preserved and what was lost during memory summarization. The SDK already captures input_content (pre-compression) and output_content (post-compression) on COMPRESS spans — this feature uses those fields to compute sentence-level semantic similarity.

Data Model

@dataclass(frozen=True)
class SentenceAnalysis:
    text: str                # the original pre-compression sentence
    best_match_score: float  # max cosine similarity to any post-content sentence
    status: str              # "preserved" (>= 0.7) or "lost" (< 0.7)

@dataclass(frozen=True)
class CompressionAudit:
    span_id: str
    semantic_loss_score: float   # 1.0 - mean(best_match_scores). 0=no loss, 1=total loss
    compression_ratio: float     # len(post_content) / len(pre_content)
    pre_sentence_count: int
    post_sentence_count: int
    sentences: list[SentenceAnalysis]
    scorer_backend: str         # "local" or "openai"

Scorer Backends

The ScorerBackend protocol requires a single method: embed(texts: list[str]) -> list[list[float]]. Three implementations are provided:

# Factory function for creating scorers:
def create_scorer(name: str) -> ScorerBackend:
    match name:
        case "local":  return LocalScorer()
        case "openai": return OpenAIScorer()
        case "mock":   return MockScorer()

Compression Analysis Algorithm

The CompressionAnalyzer.analyze(span_id, pre_content, post_content) method executes the following steps:

1. Sentence splitting: split both pre_content and post_content into sentences using a regex-based splitter (splits on .!? followed by whitespace or end-of-string, with abbreviation handling)
2. Batch embedding: concatenate all pre and post sentences into a single list and embed in one batch call — minimizing API calls for the OpenAI backend
3. Similarity matrix: for each pre-sentence, compute cosine similarity against every post-sentence using the formula: dot(a, b) / (||a|| × ||b||)
4. Best match scoring: take the maximum similarity score across all post-sentences as the best_match_score for each pre-sentence
5. Classification: score ≥ 0.7 → "preserved"; score < 0.7 → "lost"
6. Loss score: semantic_loss_score = 1.0 − mean(best_match_scores) — clamped to [0, 1]
7. Compression ratio: len(post_content) / len(pre_content) (character-level)

Loss Score Classification

Compression Audit UI View (/traces/{trace_id}/compression)

• Header: breadcrumb navigation, span metadata
• Summary card: loss score with color coding, compression ratio, preserved/lost sentence counts
• Sentence diff: list of all pre-compression sentences. Preserved sentences show a green checkmark and a score bar; lost sentences show a red X. Score bars use the same visual style as the Retrieval Debugger. Preserved sentences additionally show the best-matching post-content sentence in subtle text below
• Post-compression content: full compressed text in a code box
• "Run Audit" state: if no audit exists, a button fires POST /api/traces/{trace_id}/audit?scorer=local which runs the analysis and redirects back. The scorer parameter selects the backend (default: mock in the web UI; local or openai via CLI)

Cost Attribution — memorylens._cost

Cost attribution tracks token counts and dollar costs per memory operation. Like the Compression Auditor, it is implemented as offline enrichment — token data is captured as span attributes during instrumentation, and a CLI command computes costs using a configurable pricing model and writes cost_usd back into the span's attributes via update_span_attributes().

Capturing Token Data

Token data is added as span attributes during instrumentation, either via decorator kwargs or manually:

# Via decorator:
@instrument_write(backend="mem0", tokens_in=150, tokens_out=0, model="gpt-4o-mini")
def store(...): ...

# Via manual span creation:
with tracer.start_span(MemoryOperation.COMPRESS) as span:
    span.set_attribute("tokens_in", 1200)
    span.set_attribute("tokens_out", 300)
    span.set_attribute("model", "gpt-4o-mini")

Default Pricing Table

User Pricing Override

Users override or extend pricing via ~/.memorylens/pricing.json:

{
    "my-fine-tuned-model": {"input": 0.00001, "output": 0.00003}
}

The load_pricing() function merges user entries over defaults. User entries take precedence.

Cost Enricher

class CostEnricher:
    def enrich_span(self, attrs: dict[str, Any]) -> dict | None:
        # Returns None if no token data present
        # Returns {"cost_usd": 0.0, "_cost_warning": "..."} for unknown models
        # Returns {"cost_usd": round(cost, 10)} on success
        cost = tokens_in * pricing[model]["input"] + tokens_out * pricing[model]["output"]

update_span_attributes Mechanism

Cost enrichment writes cost_usd directly into the span's attributes JSON field in SQLite, merging without overwriting other attributes:

# SQLiteExporter method:
def update_span_attributes(self, span_id: str, new_attrs: dict) -> None:
    current = json.loads(row["attributes"])
    current.update(new_attrs)  # merge
    # UPDATE spans SET attributes = ? WHERE span_id = ?

UI Integration

No new pages are added. Cost data surfaces in existing views:

• Trace list: a "Cost" column showing $0.0012 or - if no cost data, read from span.attributes.cost_usd
• Trace detail header: cost shown after duration — 12ms · $0.0012 — only when cost_usd is present

CLI Reference — memorylens

The CLI is built with Typer and Rich. All commands read from the local SQLite database by default. Most commands support --json for machine-readable output.

memorylens init

memorylens ui

Requires memorylens[ui]. Launches web dashboard at http://127.0.0.1:8000.

memorylens traces

memorylens stats

Shows aggregate statistics: total spans, error rate, p50/p95/p99 duration, breakdown by operation and status.

memorylens config

memorylens audit

memorylens cost

Cost Report Output Format

Cost Report (grouped by operation)

OPERATION         SPANS   TOKENS IN   TOKENS OUT   TOTAL COST
memory.write        45      12,300        0          $0.0018
memory.read         38       8,400      2,100        $0.0031
memory.compress     12       6,800      1,200        $0.0015

Total: 95 spans, 27,500 tokens in, 3,300 tokens out, $0.0064

Audit Output Format

Analyzing 12 COMPRESS spans...
[████████████████████████████████] 12/12

Results:
SPAN ID      LOSS SCORE   RATIO   PRESERVED   LOST    STATUS
a1b2c3d4     0.12         0.35    8/9         1/9     ✓ low loss
e5f6g7h8     0.45         0.28    4/7         3/7     ⚠ moderate loss
i9j0k1l2     0.78         0.15    2/8         6/8     ✗ high loss

Summary: 12 spans audited. 2 with moderate loss, 1 with high loss.

Public API Reference

MemoryLens exposes exactly 8 public symbols from the top-level memorylens package. Everything else is considered internal (prefixed with _) and may change without notice.

from memorylens import (
    init, shutdown,
    instrument_write, instrument_read, instrument_compress, instrument_update,
    context, get_tracer,
)

@instrument_write(backend="mem0", capture_content=True)
def store_user_preference(user_id: str, content: str) -> bool: ...

@instrument_read(backend="mem0", top_k=5, threshold=0.7)
def search_memories(query: str) -> list[Memory]: ...

@instrument_compress(model="gpt-4o-mini")
def summarize_conversation(messages: list[str]) -> str: ...

with memorylens.context(agent_id="bot", session_id="sess-1", user_id="u-42"):
    store_memory(content)
    results = search_memories(query)
# All spans above have agent_id, session_id, user_id set

tracer = memorylens.get_tracer("my.module")
with tracer.start_span(MemoryOperation.WRITE) as span:
    span.set_attribute("memory_key", "user_prefs")
    span.set_content(input_content=content)
    result = do_write(content)
    span.set_content(output_content=str(result))

Environment Variable Overrides

Configuration

pyproject.toml

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "memorylens"
version = "0.1.0"
description = "Observability and debugging for AI agent memory systems"
requires-python = ">=3.10"
license = "Apache-2.0"

# Core dependencies (always installed)
dependencies = [
    "opentelemetry-api>=1.20",
    "opentelemetry-sdk>=1.20",
    "opentelemetry-exporter-otlp-proto-grpc>=1.20",
    "opentelemetry-exporter-otlp-proto-http>=1.20",
    "typer>=0.9",
    "rich>=13.0",
]

# Optional extras
[project.optional-dependencies]
langchain  = ["langchain-core>=0.1"]
mem0       = ["mem0ai>=0.1"]
llamaindex = ["llama-index-core>=0.10"]
letta      = ["letta-client>=0.1"]
zep        = ["zep-python>=2.0"]
ui         = ["fastapi>=0.110", "uvicorn[standard]>=0.29", "jinja2>=3.1"]
audit      = ["sentence-transformers>=2.0", "numpy>=1.24"]
dev        = ["pytest>=8.0", "pytest-asyncio>=0.23", "ruff>=0.4",
              "mypy>=1.10", "httpx>=0.27", "numpy>=1.24"]

[project.scripts]
memorylens = "memorylens.cli.main:app"

# Tooling configuration
[tool.hatch.build.targets.wheel]
packages = ["src/memorylens"]

[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]

[tool.ruff]
target-version = "py310"
line-length = 100

[tool.mypy]
python_version = "3.10"
strict = true

Optional Extras Summary

Multiple extras can be combined: pip install memorylens[langchain,mem0,ui,audit]

Local Configuration Files

MemoryLens stores local configuration in ~/.memorylens/:

Testing

MemoryLens has a comprehensive test suite with 174 tests across all layers of the system. Tests are organized to mirror the package structure.

Test Structure

Testing Patterns

Framework Integration Mocks

Integration tests use lightweight fake framework classes rather than installing the full framework. The instrumentor's _get_*_class() function is monkeypatched to return the mock class:

# Example: testing LangChain integration without installing langchain
class FakeBaseMemory:
    def save_context(self, inputs, outputs): ...
    def load_memory_variables(self, inputs): return {}

def test_langchain_write_span(monkeypatch):
    monkeypatch.setattr("memorylens.integrations.langchain.instrumentor._get_base_memory",
                        lambda: FakeBaseMemory)
    inst = LangChainInstrumentor()
    inst.instrument()
    # verify spans emitted with correct operation type

In-Memory SQLite

Database tests use SQLite in-memory connections (db_path=":memory:") for isolation and speed. Each test gets a fresh database via pytest fixtures.

Slow Test Marker

Integration tests that load the real LocalScorer (requires downloading the ~80MB all-MiniLM-L6-v2 model) are marked with @pytest.mark.slow and excluded from the default test run:

# Run fast tests only (default CI):
uv run pytest tests/ -v

# Run slow tests only (model integration):
uv run pytest tests/ -v -m slow

MockScorer for Audit Tests

The MockScorer uses MD5 hashing to produce deterministic, fast embeddings. Texts that share more words get more similar vectors, making it useful for verifying the overall analysis pipeline without requiring a real embedding model.

UI Tests with httpx

FastAPI endpoint tests use httpx.AsyncClient with an in-process test client. htmx partial responses are tested by including the HX-Request: true header. No browser automation is used — the server-driven htmx approach means API response testing fully covers all interactive behaviors.

Running Tests

# Install dev dependencies
uv sync --extra dev

# Run all fast tests
uv run pytest tests/ -v

# Run with coverage
uv run pytest tests/ --cov=memorylens --cov-report=term-missing

# Lint
uv run ruff check src/ tests/

# Type check
uv run mypy src/memorylens/

Technology Stack