symfonic.agent.factory

factory

HMSFactory -- stateless builder for the Hierarchical Memory System.

Constructs a MemoryOrchestrator (the five-layer Pentad memory stack) from backends, providers, and configuration. No side effects -- purely constructive.

HMSFactory

Stateless factory for constructing a MemoryOrchestrator.

Two entry points: - build() for custom backends - build_in_memory() for quick in-memory setup (testing/prototyping)

build classmethod

build(
    graph_backend: GraphBackend,
    vector_backend: VectorBackend,
    embedding_provider: EmbeddingProvider,
    config: OrchestratorConfig | None = None,
    *,
    stm_summary_mode: STMSummaryMode = "off",
) -> MemoryOrchestrator

Build a MemoryOrchestrator from custom backends.

Parameters:

Name	Type	Description	Default
graph_backend	GraphBackend	Persistent graph storage implementation.	required
vector_backend	VectorBackend	Vector similarity search implementation.	required
embedding_provider	EmbeddingProvider	Text-to-embedding implementation.	required
config	OrchestratorConfig | None	Orchestrator configuration. Defaults to OrchestratorConfig().	None
stm_summary_mode	STMSummaryMode	v7.0.1 R3 — threaded through to the WorkingLayer constructor so direct-API callers and engine-wired callers share the same behaviour. "off" (default) preserves v6.x byte-for-byte.	'off'

Returns:

Type	Description
MemoryOrchestrator	Fully wired MemoryOrchestrator with all five layers.

Source code in src/symfonic/agent/factory.py

@classmethod
def build(
    cls,
    graph_backend: GraphBackend,
    vector_backend: VectorBackend,
    embedding_provider: EmbeddingProvider,
    config: OrchestratorConfig | None = None,
    *,
    stm_summary_mode: STMSummaryMode = "off",
) -> MemoryOrchestrator:
    """Build a MemoryOrchestrator from custom backends.

    Args:
        graph_backend: Persistent graph storage implementation.
        vector_backend: Vector similarity search implementation.
        embedding_provider: Text-to-embedding implementation.
        config: Orchestrator configuration. Defaults to ``OrchestratorConfig()``.
        stm_summary_mode: v7.0.1 R3 — threaded through to the
            ``WorkingLayer`` constructor so direct-API callers and
            engine-wired callers share the same behaviour. ``"off"``
            (default) preserves v6.x byte-for-byte.

    Returns:
        Fully wired MemoryOrchestrator with all five layers.
    """
    cfg = config or OrchestratorConfig()

    # v7.3 Item 13.1: a single shared cache instance is handed to every
    # graph-backed layer AND to the GraphMemoryStore so identical text
    # written through different layers (or by consolidation phases that
    # bypass the layers, e.g. the Item 12 EntityLinker which calls
    # graph_store.add_node directly) reuses one embedding. Empty when
    # no provider is supplied -- maybe_embed short-circuits before
    # touching it in that case.
    embedding_cache = EmbeddingCache() if embedding_provider is not None else None
    graph_store = GraphMemoryStore(
        backend=graph_backend,
        embedding_provider=embedding_provider,
        embedding_cache=embedding_cache,
    )

    layers = cls._build_layers(
        graph_store=graph_store,
        vector_backend=vector_backend,
        embedding_provider=embedding_provider,
        embedding_cache=embedding_cache,
        enabled=cfg.enabled_layers,
        stm_summary_mode=stm_summary_mode,
        # Thread working-memory bounds so the agent path (HMSFactory)
        # honours the SAME retention config as the Orchestrator path
        # (orchestrator/factory.py). Defaults match WorkingLayer's own
        # (max_entries=10, retention=None) so default config is unchanged.
        working_max_entries=cfg.context_budget.max_history_messages,
        working_graph_retention=getattr(cfg, "working_graph_retention", None),
    )

    # v7.26.1 Item B: wire the RetrievalEngine so
    # ``FrameworkConfig.scorer_on_hot_path=True`` can route
    # ``_hydrate_impl`` through the composed scorer path.  Pre-
    # v7.26.1 ``HMSFactory.build`` produced an orchestrator without
    # a retrieval engine and the opt-in knob did not exist; both
    # behaviours are preserved when the knob stays at its False
    # default (the engine is constructed but unused).
    retrieval_engine = RetrievalEngine(
        graph_store=graph_store,
        vector_backend=vector_backend,
        scorer=RetrievalScorer(
            cfg.scoring_weights, scope_blend=cfg.scope_blend,
        ),
        traversal=GraphTraversal(graph_store),
        config=cfg,
    )

    return MemoryOrchestrator(
        layers=layers,
        graph_backend=graph_backend,
        vector_backend=vector_backend,
        embedding_provider=embedding_provider,
        embedding_cache=embedding_cache,
        config=cfg,
        retrieval_engine=retrieval_engine,
    )

build_in_memory classmethod

build_in_memory(
    embedding_provider: EmbeddingProvider,
    config: OrchestratorConfig | None = None,
    *,
    stm_summary_mode: STMSummaryMode = "off",
) -> MemoryOrchestrator

Build a MemoryOrchestrator backed entirely by in-memory stores.

Useful for testing and local prototyping. No external services needed.

Parameters:

Name	Type	Description	Default
embedding_provider	EmbeddingProvider	Text-to-embedding implementation.	required
config	OrchestratorConfig | None	Optional orchestrator configuration.	None
stm_summary_mode	STMSummaryMode	v7.0.1 R3 — threaded to WorkingLayer.	'off'

Returns:

Type	Description
MemoryOrchestrator	In-memory MemoryOrchestrator with all five layers.

Source code in src/symfonic/agent/factory.py

@classmethod
def build_in_memory(
    cls,
    embedding_provider: EmbeddingProvider,
    config: OrchestratorConfig | None = None,
    *,
    stm_summary_mode: STMSummaryMode = "off",
) -> MemoryOrchestrator:
    """Build a MemoryOrchestrator backed entirely by in-memory stores.

    Useful for testing and local prototyping.  No external services needed.

    Args:
        embedding_provider: Text-to-embedding implementation.
        config: Optional orchestrator configuration.
        stm_summary_mode: v7.0.1 R3 — threaded to ``WorkingLayer``.

    Returns:
        In-memory MemoryOrchestrator with all five layers.
    """
    return cls.build(
        graph_backend=InMemoryGraphBackend(),
        vector_backend=InMemoryVectorBackend(),
        embedding_provider=embedding_provider,
        config=config,
        stm_summary_mode=stm_summary_mode,
    )

MemoryOrchestrator dataclass

MemoryOrchestrator(
    layers: dict[MemoryLayer, BaseMemoryStore] = dict(),
    graph_backend: Any = None,
    vector_backend: Any = None,
    embedding_provider: Any = None,
    config: OrchestratorConfig = OrchestratorConfig(),
    embedding_cache: EmbeddingCache | None = None,
    retrieval_engine: RetrievalEngine | None = None,
)

Facade aggregating the five Pentad memory layers.

Holds references to all layers, backends, and the shared config. Provides a uniform get_layer() accessor and all_layers() listing.

layer_count property

layer_count: int

Number of registered layers.

all_layers

all_layers() -> dict[MemoryLayer, BaseMemoryStore]

Return all registered layers.

Source code in src/symfonic/agent/factory.py

def all_layers(self) -> dict[MemoryLayer, BaseMemoryStore]:
    """Return all registered layers."""
    return dict(self.layers)

get_layer

get_layer(layer: MemoryLayer) -> BaseMemoryStore | None

Return the store for a specific layer, or None if not registered.

Source code in src/symfonic/agent/factory.py

def get_layer(self, layer: MemoryLayer) -> BaseMemoryStore | None:
    """Return the store for a specific layer, or None if not registered."""
    return self.layers.get(layer)