teff.memory.extract

teff.memory.extract

LLM-based fact extraction for long-term memory.

The :class:MemoryExtractor turns a conversation into durable facts by asking a model to summarise what should be remembered beyond the current session. It is a thin layer over a :class:~teff.harness.loop.Harness:

• extract calls the model once and parses a JSON array of facts from the reply (tolerating code fences and surrounding prose).
• save extracts facts and writes them into a :class:~teff.memory.base.MemoryStore, keyed by a stable hash of the fact text so re-extracting the same fact upserts it instead of duplicating it.

The extractor never stores anything itself; pass it a :class:MemoryStore (which owns the vector store / embedder) to persist results.

Classes:

Name	Description
MemoryExtractor	Extract durable facts from a conversation using an LLM.

MemoryExtractor

Extract durable facts from a conversation using an LLM.

Parameters:

Name	Type	Description	Default
harness	Harness | None	A :class:~teff.harness.loop.Harness used for the single extraction call. When omitted, one is built from model and provider.	None
model	str | None	Model name for a self-built harness (ignored when harness is given).	None
provider	str | None	Provider key for a self-built harness.	None
system_prompt	str | None	Overrides the default extraction prompt.	None
temperature	float	Sampling temperature for the extraction call.	0.0

Methods:

Name	Description
extract	Return the durable facts found in conversation.
save	Extract facts and write them into memory.

Source code in teff/memory/extract.py

class MemoryExtractor:
    """Extract durable facts from a conversation using an LLM.

    Args:
        harness: A :class:`~teff.harness.loop.Harness` used for the single
            extraction call.  When omitted, one is built from *model* and
            *provider*.
        model: Model name for a self-built harness (ignored when *harness*
            is given).
        provider: Provider key for a self-built harness.
        system_prompt: Overrides the default extraction prompt.
        temperature: Sampling temperature for the extraction call.
    """

    def __init__(
        self,
        harness: Harness | None = None,
        *,
        model: str | None = None,
        provider: str | None = None,
        system_prompt: str | None = None,
        temperature: float = 0.0,
    ):
        if harness is None:
            if not model:
                raise ValueError(
                    "MemoryExtractor needs a harness or a model to build one"
                )
            harness = Harness(model=model, provider=provider, temperature=temperature)
        self._harness = harness
        self._system_prompt = system_prompt or DEFAULT_SYSTEM_PROMPT

    async def extract(self, conversation: list[dict]) -> list[str]:
        """Return the durable facts found in *conversation*.

        Args:
            conversation: OpenAI-style messages (``{"role", "content"}``);
                the text of every message is included in the prompt.
        """
        transcript = _transcript(conversation)
        reply = await self._harness.call(
            [
                {"role": "system", "content": self._system_prompt},
                {"role": "user", "content": transcript},
            ]
        )
        return _parse_facts(reply.content)

    async def save(
        self,
        memory: Any,
        conversation: list[dict],
        namespace: tuple[str, ...],
        *,
        ttl: float | None = None,
    ) -> list[tuple[str, str]]:
        """Extract facts and write them into *memory*.

        Each fact is stored under a stable key derived from its text (a
        short SHA-1), so re-extracting the same fact updates it in place.

        Args:
            memory: A :class:`~teff.memory.base.MemoryStore`.
            conversation: The messages to extract facts from.
            namespace: Namespace to store the facts under.
            ttl: Per-item TTL in seconds, or ``None`` for no expiry.

        Returns:
            The ``(key, fact)`` pairs that were written.
        """
        written: list[tuple[str, str]] = []
        for fact in await self.extract(conversation):
            key = _fact_key(fact)
            await memory.put(
                namespace, key, {"text": fact, "source": "extractor"}, ttl=ttl
            )
            written.append((key, fact))
        return written

extract async

extract(conversation)

Return the durable facts found in conversation.

Parameters:

Name	Type	Description	Default
conversation	list[dict]	OpenAI-style messages ({"role", "content"}); the text of every message is included in the prompt.	required

Source code in teff/memory/extract.py

async def extract(self, conversation: list[dict]) -> list[str]:
    """Return the durable facts found in *conversation*.

    Args:
        conversation: OpenAI-style messages (``{"role", "content"}``);
            the text of every message is included in the prompt.
    """
    transcript = _transcript(conversation)
    reply = await self._harness.call(
        [
            {"role": "system", "content": self._system_prompt},
            {"role": "user", "content": transcript},
        ]
    )
    return _parse_facts(reply.content)

save async

save(memory, conversation, namespace, *, ttl=None)

Extract facts and write them into memory.

Each fact is stored under a stable key derived from its text (a short SHA-1), so re-extracting the same fact updates it in place.

Parameters:

Name	Type	Description	Default
memory	Any	A :class:~teff.memory.base.MemoryStore.	required
conversation	list[dict]	The messages to extract facts from.	required
namespace	tuple[str, ...]	Namespace to store the facts under.	required
ttl	float | None	Per-item TTL in seconds, or None for no expiry.	None

Returns:

Type	Description
list[tuple[str, str]]	The (key, fact) pairs that were written.

Source code in teff/memory/extract.py

async def save(
    self,
    memory: Any,
    conversation: list[dict],
    namespace: tuple[str, ...],
    *,
    ttl: float | None = None,
) -> list[tuple[str, str]]:
    """Extract facts and write them into *memory*.

    Each fact is stored under a stable key derived from its text (a
    short SHA-1), so re-extracting the same fact updates it in place.

    Args:
        memory: A :class:`~teff.memory.base.MemoryStore`.
        conversation: The messages to extract facts from.
        namespace: Namespace to store the facts under.
        ttl: Per-item TTL in seconds, or ``None`` for no expiry.

    Returns:
        The ``(key, fact)`` pairs that were written.
    """
    written: list[tuple[str, str]] = []
    for fact in await self.extract(conversation):
        key = _fact_key(fact)
        await memory.put(
            namespace, key, {"text": fact, "source": "extractor"}, ttl=ttl
        )
        written.append((key, fact))
    return written