API reference: llmslim.memory
Memory, its stores and its result types, with their signatures and docstrings.
About this reference
Each entry is a name you import from llmslim.memory: agent memory with no LLM in the loop: Memory, its stores and its result types. It shows the full signature with its type annotations, and the docstring. The page is generated from the main branch (llmslim 0.12.0 and the changes since, not released yet) by scripts/api_reference.py, so it changes only when the code does.
ExplainedCandidate
One candidate a search ranked: a stored turn, a part of one, or a statement for a section.
class ExplainedCandidate:
"""One candidate a search ranked: a stored turn, a part of one, or a statement for a section.
``pool`` is what it competed in: ``"user"`` (user turns), ``"reply"`` (replies, and system and tool turns),
``"detail"`` (reply passages kept whole in a summary), or a section (``"requests"``, ``"preferences"``,
``"conflicts"``). ``kind`` is ``"turn"`` (a whole user turn), ``"window"`` (a few sentences of a reply, or a code
block or table), ``"opening"`` (a turn's opening in a summary) or ``"statement"`` (a sentence for a section).
``part`` is the window's position in its turn, or the statement's position among the turn's statements; 0
otherwise.
``rank`` is its place in its pool, 1 first, in the order selection tried the pool; ``score`` is the relevance it
was ranked by. Scores compare within a pool only. Both are ``None`` when nothing was ranked (a summary whose
openings all fit). ``tokens`` is the candidate's own size, without its label or bullet.
``status`` is ``"kept"``, ``"clipped"`` (kept in part) or ``"dropped"``; ``reason`` says why it was clipped or
dropped (see ``docs/MEMORY.md``). ``side`` is ``"denies"`` or ``"affirms"`` for a statement the conflicts section
weighed, else ``None``.
"""
turn_id: str
pool: str
kind: str
part: int
rank: Optional[int]
score: Optional[float]
tokens: int
status: str
reason: Optional[str] = None
side: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
...
@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> ExplainedCandidate:
"""The candidate ``to_dict`` gave, read back (for example from JSON)."""ExplainedSection
What a search did with one of the sections shown above the turns: "requests" (standing requests), "preferences" or "conflicts".
class ExplainedSection:
"""What a search did with one of the sections shown above the turns: ``"requests"`` (standing requests),
``"preferences"`` or ``"conflicts"``.
``shown`` says whether it is in ``MemoryContext.text``; when it is not, ``reason`` says why (see
``docs/MEMORY.md``). ``cap`` is the most of the budget its statements may take (its best one may take more on a
small budget), 0 when it was not built. ``tokens`` is what it takes in the text, its header and the blank line
after it included, and ``share`` is ``tokens`` over the budget. ``candidates`` are the statements it weighed,
best first.
"""
name: str
shown: bool
reason: Optional[str]
cap: int
tokens: int
share: float
candidates: Tuple[ExplainedCandidate, ...] = ()
def to_dict(self) -> Dict[str, Any]:
...
@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> ExplainedSection:
"""The section ``to_dict`` gave, read back (for example from JSON)."""ExplainedTurn
What a search did with one stored turn of its scope.
class ExplainedTurn:
"""What a search did with one stored turn of its scope.
``session`` is the index of the turn's session among the sessions read (oldest first), ``None`` for a turn
outside the time window. ``pool`` is ``"user"`` or ``"reply"`` when the turns were selected from, ``None`` when the
whole history fit or the turn was outside the window.
``tokens`` is what the turn's block takes in ``MemoryContext.text``, counted with the line break after it, and
``share`` is ``tokens`` over the budget: 0 for a dropped turn. ``status`` and ``reason`` are as on
:class:`ExplainedCandidate`; for a dropped reply, ``reason`` is that of its best-ranked window, and
``candidates`` holds every window with its own.
"""
id: str
role: str
created_at: datetime
session: Optional[int]
pool: Optional[str]
status: str
reason: Optional[str]
tokens: int
share: float
candidates: Tuple[ExplainedCandidate, ...] = ()
def to_dict(self) -> Dict[str, Any]:
...
@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> ExplainedTurn:
"""The turn ``to_dict`` gave, read back (for example from JSON)."""HistoryImportError
A history file that can't be imported: missing or unreadable, of an unknown format, over a size limit, an unsafe zip archive, or (with strict=True) a malformed record. The message says which file and why.
class HistoryImportError(ValueError):
"""A history file that can't be imported: missing or unreadable, of an unknown format, over a size limit, an
unsafe zip archive, or (with ``strict=True``) a malformed record. The message says which file and why.
"""ImportLimits
Bounds on what import_history reads, so a hostile or broken file fails clearly instead of exhausting memory or the stack. Every input is treated as untrusted.
class ImportLimits:
"""Bounds on what :func:`~llmslim.memory.import_history` reads, so a hostile or broken file fails clearly
instead of exhausting memory or the stack. Every input is treated as untrusted.
Args:
max_file_bytes: Largest file read, in bytes: a JSONL file, a ``conversations.json`` or a Mem0 export, and,
inside a zip archive, the member read (its uncompressed size).
max_record_bytes: Largest single record, in characters: one JSONL line, or one element of a JSON array
(one ChatGPT or Claude.ai conversation, one Mem0 memory). A larger one is skipped and counted.
max_depth: Deepest nesting of arrays and objects in one record. A deeper record is skipped and counted
before it is parsed, so a nesting bomb can't exhaust the parser.
max_turn_chars: Longest turn stored, in characters. A longer one is skipped and counted.
max_zip_members: Most entries a zip archive may list.
max_zip_ratio: Highest compression ratio (uncompressed over compressed size) of the zip member read, when it
is over 1 MiB uncompressed. A zip bomb is far above it; JSON exports compress about 5 to 20 times.
"""
max_file_bytes: int = 1 << 30
max_record_bytes: int = 32 * _MIB
max_depth: int = 64
max_turn_chars: int = 100000
max_zip_members: int = 100000
max_zip_ratio: float = 100.0ImportReport
What import_history did.
class ImportReport:
"""What :func:`~llmslim.memory.import_history` did.
``imported`` counts the turns stored (in a dry run, the turns that would be), ``duplicates`` the turns already
in memory from an earlier import, ``skipped`` the turns and records left out by reason (``thinking``,
``tool_output``, ``meta``, ``system``, ``attachment``, ``other_branch``, ``subagent``, ``unsupported``,
``empty``, ``outside_window``, ``too_large``, ``too_deep``), and ``malformed`` the lines or records that were not
valid JSON or not in the format's shape. ``redactions`` counts the credentials replaced, by kind. ``errors``
lists the files that could not be read at all; the other files were still imported. ``limited`` is true when
the import stopped at ``limit``.
"""
format: str
source: str
dry_run: bool
files: int
sessions: Tuple[ImportedSession, ...]
imported: int
duplicates: int
malformed: int
skipped: Mapping[str, int] = field(default_factory=dict)
redactions: Mapping[str, int] = field(default_factory=dict)
errors: Tuple[str, ...] = ()
limited: bool = False
@property
def ok(self) -> bool:
"""True when every file was read: no entry in ``errors``."""
@property
def skipped_total(self) -> int:
"""All the turns and records left out, over every reason."""
def to_dict(self) -> Dict[str, Any]:
"""The report as plain JSON-ready data (what ``llmslim import --json`` prints)."""ImportedSession
One session (a Claude Code transcript, a ChatGPT or Claude.ai conversation, or one scope of a Mem0 export or JSONL file) as import_history imported it.
class ImportedSession:
"""One session (a Claude Code transcript, a ChatGPT or Claude.ai conversation, or one scope of a Mem0 export or
JSONL file) as :func:`~llmslim.memory.import_history` imported it.
``turns`` counts the turns found to import; ``imported`` the new ones stored (or, in a dry run, that would be),
``duplicates`` the ones already in memory, and ``skipped`` the ones left out here (outside ``since``/``until``,
over ``max_turn_chars``, or empty once credentials were redacted).
"""
format: str
session_id: str
title: Optional[str]
run_id: Optional[str]
user_id: str
turns: int
imported: int
duplicates: int
skipped: int
def to_dict(self) -> Dict[str, Any]:
"""The session as plain JSON-ready data."""InMemoryStore
Turns in a list in this process. Nothing survives a restart.
class InMemoryStore:
"""Turns in a list in this process. Nothing survives a restart."""
def __init__(self) -> None:
...
def add(self, items: Sequence[MemoryItem]) -> None:
...
def list(
self,
scope: Scope,
*,
since: Optional[datetime] = None,
until: Optional[datetime] = None,
) -> List[MemoryItem]:
...
def get(self, item_id: str) -> Optional[MemoryItem]:
...
def update(
self,
item_id: str,
*,
content: Optional[str] = None,
metadata: Optional[Mapping[str, Any]] = None,
) -> Optional[MemoryItem]:
...
def delete(self, item_id: str) -> bool:
...
def delete_scope(self, scope: Scope) -> int:
...Memory
Long-term memory for an agent that keeps every turn verbatim and selects at read time.
class Memory:
"""Long-term memory for an agent that keeps every turn verbatim and selects at read time.
Nothing is summarized or extracted on the way in, so writes cost no model call and a stored turn never turns into
a wrong "fact". On the way out, :meth:`search` ranks the stored turns against the question and returns what
answers it within a token budget: the user's own turns (whole where they fit) and the most relevant parts of the
replies, one labelled block per turn under its date, in date order. No LLM is called to write or to read (by
default no model runs at all; ``relevance="semantic"`` adds a small local cross-encoder), and on one installation the same
history and question give the same text (see ``docs/MEMORY.md``, Limits).
Args:
store: Where turns live: a :class:`~llmslim.memory.MemoryStore`, a path to a SQLite file, a
``postgresql://`` URL (:class:`~llmslim.memory.PostgresStore`), a ``mongodb://`` or ``mongodb+srv://`` URL
(:class:`~llmslim.memory.MongoStore`), or ``None`` for an in-process store.
budget: Default token budget for :meth:`search`.
relevance: Default relevance for :meth:`search`: ``"lexical"`` (no model), ``"semantic"`` (LLMSlim's optional
22M cross-encoder, ``pip install "llmslim[semantic]"``) or a ``scorer(query, texts)`` callable.
tz: Time zone for the turn dates shown to the model. Defaults to UTC.
stamp_format: strftime format of those dates. The default reads like 2026-10-17 (Sat) 14:05.
date_format: What a kept turn shows, when a long history is selected from, if its time is the one shown last
in its session (the first kept turn of a session, and a turn whose time differs, show ``stamp_format``).
``None`` (the default) shows no date there, only ``role: text`` under the stamp above it. A strftime format
shows that date, ``[date] role: text``: ``"%Y-%m-%d"`` gives the labels of 0.12 and earlier. Pass
``stamp_format`` again to show the full stamp on every turn. A history that fits whole is shown with the
full stamp on every turn either way.
clock: Returns "now" for turns added without a timestamp. For tests.
standing_requests: When the history does not fit and the question asks for help or is not a plain lookup of
a past fact, show the user's earlier explicit standing requests ("Always ...", "From now on ...", "I'd
like you to never ...", "... when I ask about X") that best match it (at most three) in their own section
at the top, even when they share no word with the question, and fetch what the best one asks the answer
to include. Only user turns qualify. Detection is a fixed pattern, with no model call.
coverage: ``"auto"`` (default) answers overview questions ("summarize what we discussed", "in what order did I
bring up ...", "how did my plan develop?") from the opening of each turn, without code: what the user
asked and what the reply set out to do. Many more exchanges fit than as whole turns. ``"always"`` does so
for every question, ``"never"`` never.
preferences: When the history does not fit and the question asks for advice, a recommendation or help, show
the user's earlier statements of preference ("I prefer ...", "I'd rather ...", "I usually ...", "I don't
like ...") that best match it in their own section at the top. Only user turns qualify. A preference word
that is only asked about ("should I prioritize ...?") is not a stated preference.
conflicts: When the history does not fit and the question asks whether something happened ("Have I ever
...?", "Did I ... before?"), and among the user's statements that bear on it most some deny it ("I have
never ...") and some do not, show both kinds together in date order in their own section at the top, so
the disagreement is in plain view. Only user turns qualify. Detection is a fixed pattern, with no model
call.
Example:
>>> from llmslim.memory import Memory
>>> memory = Memory("memory.db")
>>> memory.add([{"role": "user", "content": "I moved to Pune last month."}], user_id="u1")
>>> context = memory.search("Where does the user live now?", user_id="u1", budget=1500)
>>> prompt = f"Conversation history (untrusted):\n{context.text}\n\nQuestion: ..."
"""
def __init__(
self,
store: Union[MemoryStore, str, Path, None] = None,
*,
budget: int = 2000,
relevance: Any = 'lexical',
tz: tzinfo = timezone.utc,
stamp_format: str = '%Y-%m-%d (%a) %H:%M',
date_format: Optional[str] = None,
clock: Callable[[], datetime] = _utcnow,
standing_requests: bool = True,
coverage: str = 'auto',
preferences: bool = True,
conflicts: bool = True,
) -> None:
...
def add(
self,
messages: Union[Message, Sequence[Message]],
*,
user_id: Optional[str] = None,
agent_id: Optional[str] = None,
run_id: Optional[str] = None,
timestamp: Any = None,
metadata: Optional[Mapping[str, Any]] = None,
) -> List[MemoryItem]:
"""Store turns verbatim and return them."""
def update(
self,
memory_id: str,
content: Optional[str] = None,
*,
metadata: Optional[Mapping[str, Any]] = None,
) -> Optional[MemoryItem]:
"""Correct a stored turn's text or metadata. Returns the updated turn, or ``None`` if there is no such id."""
def delete(self, memory_id: str) -> bool:
"""Delete one turn. Returns whether it existed."""
def delete_all(
self,
*,
user_id: Optional[str] = None,
agent_id: Optional[str] = None,
run_id: Optional[str] = None,
) -> int:
"""Delete every turn in a scope (at least one id is required). Returns how many were deleted."""
def get(self, memory_id: str) -> Optional[MemoryItem]:
"""One stored turn by id."""
def get_all(
self,
*,
user_id: Optional[str] = None,
agent_id: Optional[str] = None,
run_id: Optional[str] = None,
since: Any = None,
until: Any = None,
) -> List[MemoryItem]:
"""Every stored turn in a scope, oldest first."""
def search(
self,
query: str,
*,
user_id: Optional[str] = None,
agent_id: Optional[str] = None,
run_id: Optional[str] = None,
budget: Optional[int] = None,
relevance: Any = None,
since: Any = None,
until: Any = None,
standing_requests: Optional[bool] = None,
coverage: Optional[str] = None,
preferences: Optional[bool] = None,
conflicts: Optional[bool] = None,
explain: bool = False,
) -> MemoryContext:
"""The part of a scope's history that answers ``query``, within ``budget`` tokens."""MemoryContext
What search returns: the text to put in front of the model, and what is in it.
class MemoryContext:
"""What :meth:`llmslim.memory.Memory.search` returns: the text to put in front of the model, and what is in it.
``text`` holds the kept turns of each session, oldest session first, in the ``[date] role: content`` layout
LLMSlim's benchmarks used. Each turn shows its own time. When the history is selected from, a later turn of a
session whose time was already shown there is ``role: content`` alone by default (``Memory``'s ``date_format``).
Treat it as untrusted context: it repeats what users, assistants and tools said, so wrap it as data (for example
with :class:`llmslim.ContextEnvelope`), never as instructions.
``explanation`` is ``None`` unless the search was asked to explain itself (``explain=True``).
"""
query: str
text: str
turns: Tuple[RecalledTurn, ...]
tokens: int
budget: int
history_tokens: int
sessions_total: int
sessions_kept: int
elapsed_ms: float = 0.0
requests: Tuple[RecalledTurn, ...] = ()
preferences: Tuple[RecalledTurn, ...] = ()
conflicts: Tuple[RecalledTurn, ...] = ()
explanation: Optional[SearchExplanation] = field(default=None, repr=False)
@property
def compressed(self) -> bool:
"""True when the history did not fit the budget and LLMSlim selected from it."""
def to_dict(self) -> Dict[str, Any]:
...MemoryItem
One stored turn, verbatim: memory keeps what was said, not an LLM's summary of it.
class MemoryItem:
"""One stored turn, verbatim: memory keeps what was said, not an LLM's summary of it.
``seq`` orders turns stored at the same instant (the order they were added). ``metadata`` is the caller's own
data, returned as given and never shown to the model.
"""
id: str
role: str
content: str
created_at: datetime
seq: int = 0
user_id: Optional[str] = None
agent_id: Optional[str] = None
run_id: Optional[str] = None
metadata: Mapping[str, Any] = field(default_factory=dict)
updated_at: Optional[datetime] = None
def to_dict(self) -> Dict[str, Any]:
...MemoryStore
Persistence for memory turns. Implementations must be safe to call from several threads.
class MemoryStore(Protocol):
"""Persistence for memory turns. Implementations must be safe to call from several threads."""
def add(self, items: Sequence[MemoryItem]) -> None:
"""Store new turns. Their ``id`` values are new; their ``seq`` values are ignored and assigned by the store."""
def list(
self,
scope: Scope,
*,
since: Optional[datetime] = None,
until: Optional[datetime] = None,
) -> List[MemoryItem]:
"""Turns in ``scope`` with ``since <= created_at < until``, oldest first (ties in the order they were added)."""
def get(self, item_id: str) -> Optional[MemoryItem]:
"""One turn by id, or ``None``."""
def update(
self,
item_id: str,
*,
content: Optional[str] = None,
metadata: Optional[Mapping[str, Any]] = None,
) -> Optional[MemoryItem]:
"""Replace a turn's content and/or metadata; return the updated turn, or ``None`` if there is no such id."""
def delete(self, item_id: str) -> bool:
"""Delete one turn; return whether it existed."""
def delete_scope(self, scope: Scope) -> int:
"""Delete every turn in ``scope``; return how many were deleted."""MongoStore
Turns in one MongoDB collection.
class MongoStore:
"""Turns in one MongoDB collection.
Pass a connection URL (``mongodb://user:password@host:27017/db``, or ``mongodb+srv://...`` for Atlas), in which
case the store opens its own ``MongoClient`` and closes it in :meth:`close`; a ``pymongo.MongoClient``; or a
``pymongo.database.Database``. The store is safe to use from several threads, and several processes can share one
collection.
The collection (``llmslim_memory`` by default), its indexes and a ``<collection>_meta`` collection, which holds
the schema version and the counter that orders turns, are created when the store opens if they don't exist.
Creating them again changes nothing, so several processes can start at once. A collection written by a newer
llmslim is refused rather than misread. On a replica set or a sharded cluster each :meth:`add` is one transaction,
so a batch is stored whole or not at all, even if the connection drops partway. MongoDB aborts a transaction that
runs longer than 60 seconds by default, so add a long history in batches (``llmslim import`` writes at most 500
turns or about 4 MB at a time). A standalone server has no transactions; there a batch that fails partway is undone
instead. Reads go to the primary, so a turn is readable as soon as it is added.
Every caller value is stored and matched as data: ids and scope values must be strings and are compared with
``$eq``, and metadata whose keys MongoDB would read as operators or paths (a leading ``$``, a ``.``) is kept as JSON
text. Error messages from opening the store have the URL's password and secret options removed.
Args:
client: A ``mongodb://`` or ``mongodb+srv://`` URL, a ``MongoClient`` or a ``Database``.
database: Database name: letters, digits, ``_`` and ``-``, and not ``admin``, ``local`` or ``config``. By
default the URL's (or the client's) default database, else ``llmslim``. Leave it unset with a
``Database``.
collection: Collection name: letters, digits and underscores. Use one per application or tenant if you like.
"""
def __init__(
self,
client: Any,
*,
database: Optional[str] = None,
collection: str = 'llmslim_memory',
) -> None:
...
def close(self) -> None:
"""Close the client the store opened for a URL. A client or database you passed in stays open."""
def add(self, items: Sequence[MemoryItem]) -> None:
...
def list(
self,
scope: Scope,
*,
since: Optional[datetime] = None,
until: Optional[datetime] = None,
) -> List[MemoryItem]:
...
def get(self, item_id: str) -> Optional[MemoryItem]:
...
def update(
self,
item_id: str,
*,
content: Optional[str] = None,
metadata: Optional[Mapping[str, Any]] = None,
) -> Optional[MemoryItem]:
...
def delete(self, item_id: str) -> bool:
...
def delete_scope(self, scope: Scope) -> int:
...PostgresStore
Turns in one Postgres table.
class PostgresStore:
"""Turns in one Postgres table.
Pass a connection URL (``postgresql://user:password@host:5432/db``), in which case the store opens its own
connection pool and closes it in :meth:`close`; a ``psycopg_pool.ConnectionPool``; or a single
``psycopg.Connection`` in autocommit mode. The store is safe to use from several threads.
The table (``llmslim_memory`` by default) and its ``<table>_meta`` table are created if they don't exist. Creation
runs under an advisory lock, so several processes can start at once. The schema is versioned: a table written by
a newer llmslim is refused rather than misread.
Args:
conninfo: A connection URL, a ``ConnectionPool`` or an autocommit ``Connection``.
table: Table name: letters, digits and underscores. Use one per application or tenant if you like.
max_size: Most connections in the pool the store opens for a URL.
"""
def __init__(self, conninfo: Any, *, table: str = 'llmslim_memory', max_size: int = 10) -> None:
...
def close(self) -> None:
"""Close the pool the store opened for a URL. A pool or connection you passed in stays open."""
def add(self, items: Sequence[MemoryItem]) -> None:
...
def list(
self,
scope: Scope,
*,
since: Optional[datetime] = None,
until: Optional[datetime] = None,
) -> List[MemoryItem]:
...
def get(self, item_id: str) -> Optional[MemoryItem]:
...
def update(
self,
item_id: str,
*,
content: Optional[str] = None,
metadata: Optional[Mapping[str, Any]] = None,
) -> Optional[MemoryItem]:
...
def delete(self, item_id: str) -> bool:
...
def delete_scope(self, scope: Scope) -> int:
...RecalledTurn
A stored turn that made it into a MemoryContext.
class RecalledTurn:
"""A stored turn that made it into a :class:`MemoryContext`.
``partial`` is true when the turn was too long to keep whole and only its most relevant sentences were kept.
``score`` is a bounded lexical overlap between the query and the turn (0 to 1), for ordering and display; the
selection itself is LLMSlim's query-focused ranking of every session's turns.
"""
item: MemoryItem
score: float
partial: bool = FalseSQLiteStore
Turns in one SQLite file, through Python's built-in sqlite3.
class SQLiteStore:
"""Turns in one SQLite file, through Python's built-in ``sqlite3``.
Pass a file path, or ``":memory:"`` for a private in-memory database. One connection is shared behind a lock, so
the store is safe to use from several threads in one process. Timestamps are stored as UTC ISO-8601 text, which
sorts in time order. The schema is versioned in ``llmslim_meta``; a file written by a newer schema is refused rather
than misread.
"""
def __init__(self, path: Union[str, Path] = 'llmslim_memory.db') -> None:
...
def close(self) -> None:
...
def add(self, items: Sequence[MemoryItem]) -> None:
...
def list(
self,
scope: Scope,
*,
since: Optional[datetime] = None,
until: Optional[datetime] = None,
) -> List[MemoryItem]:
...
def get(self, item_id: str) -> Optional[MemoryItem]:
...
def update(
self,
item_id: str,
*,
content: Optional[str] = None,
metadata: Optional[Mapping[str, Any]] = None,
) -> Optional[MemoryItem]:
...
def delete(self, item_id: str) -> bool:
...
def delete_scope(self, scope: Scope) -> int:
...Scope
Whose memory an operation reads or writes, in Mem0's terms.
class Scope:
"""Whose memory an operation reads or writes, in Mem0's terms.
``user_id`` is the person, ``agent_id`` the agent, and ``run_id`` one session or run. A field left as ``None``
matches every value when reading, so ``Scope(user_id="u1")`` reads all of u1's sessions with every agent. At least
one field must be set: an empty scope would read or delete everyone's memory.
"""
user_id: Optional[str] = None
agent_id: Optional[str] = None
run_id: Optional[str] = None
def matches(
self,
user_id: Optional[str],
agent_id: Optional[str],
run_id: Optional[str],
) -> bool:
...SearchExplanation
Why search returned what it did, from what it decided: given with explain=True as MemoryContext.explanation.
class SearchExplanation:
"""Why :meth:`llmslim.memory.Memory.search` returned what it did, from what it decided: given with
``explain=True`` as ``MemoryContext.explanation``.
``mode`` is ``"whole_history"`` (it fit the budget, so nothing was selected), ``"turns"`` (turns ranked by
relevance) or ``"digest"`` (a summary from the opening of each turn). ``route`` is the kind of question
(``"lookup"``, ``"advice"``, ``"overview"`` or ``"other"``), ``None`` when the whole history fit.
``room`` is what the sections left of the budget for turns. ``user_room`` is the part of it reserved for user
turns (or for their openings in a summary) and ``user_turn_cap`` the most one user turn may take whole; both are
``None`` when they did not apply. ``clipped`` is true when the context was cut to whole words as a last resort.
``turns`` covers every stored turn in the scope, oldest first, including those outside the time window, and
``sections`` the three sections. Their ``tokens`` add up to ``tokens`` (what ``MemoryContext.text`` takes) with
``unattributed_tokens``, which is 0 with tiktoken; with the 4-characters-per-token estimate, counting the blocks
one by one can round a few tokens either way. Nothing here repeats stored text: turns are named by their ids.
"""
mode: str
route: Optional[str]
budget: int
tokens: int
history_tokens: int
room: int
user_room: Optional[int]
user_turn_cap: Optional[int]
clipped: bool
unattributed_tokens: int
turns: Tuple[ExplainedTurn, ...]
sections: Tuple[ExplainedSection, ...]
def to_dict(self) -> Dict[str, Any]:
"""The explanation as JSON-ready data; :meth:`from_dict` reads it back."""
@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> SearchExplanation:
"""The explanation ``to_dict`` gave, read back (for example from JSON)."""import_history()
Import chat history from another tool into memory, verbatim, as llmslim import does.
def import_history(
source: Union[str, os.PathLike[str]],
memory: Union[Memory, MemoryStore, str, os.PathLike[str], None],
*,
format: str = 'auto',
user_id: Optional[str] = None,
agent_id: Optional[str] = None,
since: Any = None,
until: Any = None,
limit: Optional[int] = None,
include_tool_output: bool = False,
dry_run: bool = False,
strict: bool = False,
redact: bool = True,
limits: Optional[ImportLimits] = None,
on_session: Optional[Callable[[ImportedSession], None]] = None,
) -> ImportReport:
"""Import chat history from another tool into memory, verbatim, as ``llmslim import`` does.
Reads Claude Code transcripts (``~/.claude/projects``), a ChatGPT or Claude.ai data export (the zip as
downloaded, its folder, or ``conversations.json``), a Mem0 export (``get_all`` output or a list of memories) or
generic JSONL (one ``{"role", "content", "timestamp"?, "session_id"?, "user_id"?}`` per line). Each session
becomes a ``run_id`` and each message a turn with its original time. Only what the user and the assistant said is
imported: reasoning ("thinking") never is, and tool calls and results only with ``include_tool_output``.
Importing is idempotent: each turn's id is derived from its source id (or, without one, from its content and
position in its session), so importing the same history again stores only what is new. Every file is untrusted
and read within ``limits``; see ``docs/IMPORT.md``. No model is called and nothing leaves the machine.
Args:
source: The file or folder to read. ``~`` is expanded.
memory: Where the turns go: a :class:`~llmslim.memory.Memory`, a store, a SQLite path, or a
``postgresql://`` or ``mongodb://`` URL. ``None`` only with ``dry_run``.
format: ``"claude-code"``, ``"chatgpt"``, ``"claude-ai"``, ``"mem0"``, ``"jsonl"`` or ``"auto"``, which
tells them apart from what the file holds (not its name).
user_id: The user every turn is stored under. When ``None``, a record's own ``user_id`` (Mem0, JSONL) is
kept, else ``"default"``, the user ``llmslim mcp`` serves when none is named.
agent_id: The agent every turn is stored under; when ``None``, a record's own ``agent_id``, if any.
since: Import only turns at or after this time (a datetime, epoch seconds or ISO-8601 text).
until: Import only turns before this time.
limit: Stop after this many new turns.
include_tool_output: Also import tool calls and their results, as ``tool`` turns. Off by default: tool
output (shell output, file contents) is where secrets usually are.
dry_run: Read and count everything, store nothing. A SQLite file is opened read-only, and one that does not
exist is not created.
strict: Stop with :class:`HistoryImportError` at the first malformed or oversized record, instead of
skipping and counting it.
redact: Replace obvious credentials (API keys, AWS keys, private key blocks, bearer tokens, passwords in
URLs) with ``[REDACTED:<kind>]`` before anything is stored. Best-effort; see ``docs/IMPORT.md``.
limits: Size and nesting bounds on what is read; :class:`ImportLimits` has the defaults.
on_session: Called with each :class:`ImportedSession` as it is done, for progress.
Returns:
An :class:`ImportReport` with what was imported, skipped (by reason), malformed and redacted.
Raises:
HistoryImportError: The source can't be read at all, its format can't be told, a size limit is exceeded,
a zip archive is unsafe, or (with ``strict``) a record is malformed. A file that fails part way through
is listed in the report's ``errors`` instead, with what was imported before it kept.
ValueError: An unknown ``format``, an unreadable ``since`` or ``until``, or a ``limit`` below 1.
Example:
>>> from llmslim.memory import Memory, import_history
>>> memory = Memory("memory.db")
>>> report = import_history("~/.claude/projects", memory, format="claude-code")
>>> print(report.imported, "turns imported,", report.duplicates, "already there")
"""