Skip to content

APIs

dmf.memory.api.Memory

Small public facade over the internal temporal/LTM memory runtime.

Callers should use this wrapper instead of reaching into TemporalMemory directly. The facade speaks in application terms: text queries in, prompt context or recalled candidates out.

Parameters:

Name Type Description Default
temporal_memory TemporalMemory

Active-memory runtime that owns pruning, LTM hooks, and context rendering.

 required
embedding_engine EmbeddingEngine

Embedding provider used to vectorize text queries.

 required
candidate_generation_config CandidateGenerationConfig | None

Optional structured retrieval settings.

 None
assembly_config EvidenceAssemblyConfig | None

Optional evidence assembly settings.

 None

Returns:

Type Description

Facade instance for runtime memory operations.
Source code in dmf/memory/api.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
class Memory:
    """Small public facade over the internal temporal/LTM memory runtime.

    Callers should use this wrapper instead of reaching into
    ``TemporalMemory`` directly. The facade speaks in application terms:
    text queries in, prompt context or recalled candidates out.

    Args:
        temporal_memory: Active-memory runtime that owns pruning, LTM hooks, and
            context rendering.
        embedding_engine: Embedding provider used to vectorize text queries.
        candidate_generation_config: Optional structured retrieval settings.
        assembly_config: Optional evidence assembly settings.

    Returns:
        Facade instance for runtime memory operations.

    Raises:
        None.
    """

    def __init__(
        self,
        temporal_memory: TemporalMemory,
        embedding_engine: EmbeddingEngine,
        candidate_generation_config: CandidateGenerationConfig | None = None,
        assembly_config: EvidenceAssemblyConfig | None = None,
    ) -> None:
        self._temporal_memory = temporal_memory
        self._embedding_engine = embedding_engine
        self._candidate_generation_config = candidate_generation_config
        self._assembly_config = assembly_config

    @classmethod
    def from_dmf_config(
        cls,
        config: DMFConfig,
        temporal_memory: TemporalMemory,
        embedding_engine: EmbeddingEngine,
    ) -> Memory:
        """Construct a ``Memory`` facade from a fully-parsed ``DMFConfig``.

        Args:
            config: Parsed DMF configuration tree.
            temporal_memory: Active-memory runtime to wrap.
            embedding_engine: Embedding provider used for query vectors.

        Returns:
            Fully wired memory facade with structured retrieval enabled.

        Raises:
            AttributeError: If the supplied config does not expose the expected
                ``retrieval`` settings.
        """
        from dmf.memory.candidate_generation import CandidateGenerationConfig
        from dmf.memory.evidence_assembly import EvidenceAssemblyConfig

        r = config.retrieval
        candidate_generation_config = CandidateGenerationConfig(
            card_prefetch_k=r.card_prefetch_k,
            raw_prefetch_k=r.raw_prefetch_k,
            symbolic_lookup_k=r.symbolic_lookup_k,
            enable_card_semantic=r.enable_card_semantic,
            enable_card_symbolic=r.enable_card_symbolic,
            enable_raw_semantic=r.enable_raw_semantic,
            enable_raw_lexical=r.enable_raw_lexical,
        )
        assembly_config = EvidenceAssemblyConfig(
            final_recall_limit=r.final_recall_limit,
            max_support_turns_per_card=r.max_support_turns_per_card,
            include_superseded_when_historical=r.include_superseded_when_historical,
            include_neighbor_turns=r.include_neighbor_turns,
        )
        return cls(
            temporal_memory,
            embedding_engine,
            candidate_generation_config=candidate_generation_config,
            assembly_config=assembly_config,
        )

    def retrieve(self, query_text: str) -> list[RetrievedEvidence]:
        """Return final retrieved evidence for one query text.

        Args:
            query_text: Text query to parse, embed, retrieve, and rerank.

        Returns:
            Final evidence after candidate generation, answerability reranking,
            and evidence assembly.

        Raises:
            RuntimeError: If the structured retrieval stack was not configured.
            Backend-specific exceptions may propagate from the configured LTM
            hook.
        """
        if self._candidate_generation_config is None or self._assembly_config is None:
            raise RuntimeError(
                "Memory.retrieve() requires structured stack components. "
                "Use Memory.from_dmf_config() to construct Memory with the new retrieval stack."
            )

        from dmf.memory.answerability_rerank import rerank_answerable_evidence
        from dmf.memory.candidate_generation import CandidateGenerator
        from dmf.memory.evidence_assembly import assemble_final_evidence
        from dmf.memory.query_understanding import parse_query_frame

        query_vector = self._embed_query_text(query_text)
        query_frame = parse_query_frame(
            query_text,
            analyzer=getattr(self._temporal_memory, "_nlp_engine", None),
        )
        query_frame = dataclasses.replace(
            query_frame,
            query_embedding=query_vector.tolist(),
        )

        ltm_hook = self._temporal_memory.ltm_hook
        active_raw_records = getattr(
            self._temporal_memory,
            "get_active_raw_records",
            lambda: [],
        )()
        raw_records = [
            *ltm_hook.read_all(),
            *active_raw_records,
        ]
        if hasattr(ltm_hook, "card_store") and ltm_hook.card_store is not None:
            cards = ltm_hook.card_store.read_all()
        else:
            cards = []

        generator = CandidateGenerator(
            cards=cards,
            ltm_hook=ltm_hook,
            raw_records=raw_records,
            config=self._candidate_generation_config,
        )
        pool = generator.generate(query_frame)
        reranked = rerank_answerable_evidence(query_frame, pool.candidates)
        return assemble_final_evidence(
            reranked,
            raw_records=raw_records,
            query=query_frame,
            config=self._assembly_config,
        )

    def render_context(self, query_text: str) -> str:
        """Return a rendered prompt-ready context string for one query.

        Args:
            query_text: Text query to retrieve and render.

        Returns:
            Prompt-ready structured evidence context.

        Raises:
            RuntimeError: If the structured retrieval stack was not configured.
            Backend-specific exceptions may propagate from the configured LTM
            hook.
        """
        from dmf.memory.evidence_assembly import render_evidence_context

        evidence = self.retrieve(query_text)
        return render_evidence_context(evidence)

    def _embed_query_text(self, query_text: str) -> np.ndarray:
        """Return the canonical DMF embedding for one query text."""
        return self._embedding_engine.get_embedding(query_text)

from_dmf_config(config, temporal_memory, embedding_engine) classmethod

Construct a Memory facade from a fully-parsed DMFConfig.

Parameters:

Name Type Description Default
config DMFConfig

Parsed DMF configuration tree.

 required
temporal_memory TemporalMemory

Active-memory runtime to wrap.

 required
embedding_engine EmbeddingEngine

Embedding provider used for query vectors.

 required

Returns:

Type Description
Memory

Fully wired memory facade with structured retrieval enabled.

Raises:

Type Description
AttributeError

If the supplied config does not expose the expected retrieval settings.
Source code in dmf/memory/api.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@classmethod
def from_dmf_config(
    cls,
    config: DMFConfig,
    temporal_memory: TemporalMemory,
    embedding_engine: EmbeddingEngine,
) -> Memory:
    """Construct a ``Memory`` facade from a fully-parsed ``DMFConfig``.

    Args:
        config: Parsed DMF configuration tree.
        temporal_memory: Active-memory runtime to wrap.
        embedding_engine: Embedding provider used for query vectors.

    Returns:
        Fully wired memory facade with structured retrieval enabled.

    Raises:
        AttributeError: If the supplied config does not expose the expected
            ``retrieval`` settings.
    """
    from dmf.memory.candidate_generation import CandidateGenerationConfig
    from dmf.memory.evidence_assembly import EvidenceAssemblyConfig

    r = config.retrieval
    candidate_generation_config = CandidateGenerationConfig(
        card_prefetch_k=r.card_prefetch_k,
        raw_prefetch_k=r.raw_prefetch_k,
        symbolic_lookup_k=r.symbolic_lookup_k,
        enable_card_semantic=r.enable_card_semantic,
        enable_card_symbolic=r.enable_card_symbolic,
        enable_raw_semantic=r.enable_raw_semantic,
        enable_raw_lexical=r.enable_raw_lexical,
    )
    assembly_config = EvidenceAssemblyConfig(
        final_recall_limit=r.final_recall_limit,
        max_support_turns_per_card=r.max_support_turns_per_card,
        include_superseded_when_historical=r.include_superseded_when_historical,
        include_neighbor_turns=r.include_neighbor_turns,
    )
    return cls(
        temporal_memory,
        embedding_engine,
        candidate_generation_config=candidate_generation_config,
        assembly_config=assembly_config,
    )

render_context(query_text)

Return a rendered prompt-ready context string for one query.

Parameters:

Name Type Description Default
query_text str

Text query to retrieve and render.

 required

Returns:

Type Description
str

Prompt-ready structured evidence context.

Raises:

Type Description
RuntimeError

If the structured retrieval stack was not configured.
Source code in dmf/memory/api.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def render_context(self, query_text: str) -> str:
    """Return a rendered prompt-ready context string for one query.

    Args:
        query_text: Text query to retrieve and render.

    Returns:
        Prompt-ready structured evidence context.

    Raises:
        RuntimeError: If the structured retrieval stack was not configured.
        Backend-specific exceptions may propagate from the configured LTM
        hook.
    """
    from dmf.memory.evidence_assembly import render_evidence_context

    evidence = self.retrieve(query_text)
    return render_evidence_context(evidence)

retrieve(query_text)

Return final retrieved evidence for one query text.

Parameters:

Name Type Description Default
query_text str

Text query to parse, embed, retrieve, and rerank.

 required

Returns:

Type Description
list[RetrievedEvidence]

Final evidence after candidate generation, answerability reranking,
list[RetrievedEvidence]

and evidence assembly.

Raises:

Type Description
RuntimeError

If the structured retrieval stack was not configured.
Source code in dmf/memory/api.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def retrieve(self, query_text: str) -> list[RetrievedEvidence]:
    """Return final retrieved evidence for one query text.

    Args:
        query_text: Text query to parse, embed, retrieve, and rerank.

    Returns:
        Final evidence after candidate generation, answerability reranking,
        and evidence assembly.

    Raises:
        RuntimeError: If the structured retrieval stack was not configured.
        Backend-specific exceptions may propagate from the configured LTM
        hook.
    """
    if self._candidate_generation_config is None or self._assembly_config is None:
        raise RuntimeError(
            "Memory.retrieve() requires structured stack components. "
            "Use Memory.from_dmf_config() to construct Memory with the new retrieval stack."
        )

    from dmf.memory.answerability_rerank import rerank_answerable_evidence
    from dmf.memory.candidate_generation import CandidateGenerator
    from dmf.memory.evidence_assembly import assemble_final_evidence
    from dmf.memory.query_understanding import parse_query_frame

    query_vector = self._embed_query_text(query_text)
    query_frame = parse_query_frame(
        query_text,
        analyzer=getattr(self._temporal_memory, "_nlp_engine", None),
    )
    query_frame = dataclasses.replace(
        query_frame,
        query_embedding=query_vector.tolist(),
    )

    ltm_hook = self._temporal_memory.ltm_hook
    active_raw_records = getattr(
        self._temporal_memory,
        "get_active_raw_records",
        lambda: [],
    )()
    raw_records = [
        *ltm_hook.read_all(),
        *active_raw_records,
    ]
    if hasattr(ltm_hook, "card_store") and ltm_hook.card_store is not None:
        cards = ltm_hook.card_store.read_all()
    else:
        cards = []

    generator = CandidateGenerator(
        cards=cards,
        ltm_hook=ltm_hook,
        raw_records=raw_records,
        config=self._candidate_generation_config,
    )
    pool = generator.generate(query_frame)
    reranked = rerank_answerable_evidence(query_frame, pool.candidates)
    return assemble_final_evidence(
        reranked,
        raw_records=raw_records,
        query=query_frame,
        config=self._assembly_config,
    )

options: filters: - "!^[^]" - "!^init$"