symfonic.memory.orchestrator.factory

factory

OrchestratorFactory -- DI wiring for MemoryOrchestrator.

Centralises the construction of all internal components from backends and configuration. Uses a layer registry for Open/Closed compliance: new memory layers can be registered without modifying existing code.

HybridBackendConfig dataclass

HybridBackendConfig(
    layer_graph_backends: dict[
        MemoryLayer, GraphBackend
    ] = dict(),
    layer_vector_backends: dict[
        MemoryLayer, VectorBackend
    ] = dict(),
    layer_embedding_providers: dict[
        MemoryLayer, EmbeddingProvider
    ] = dict(),
)

Per-layer backend overrides for hybrid storage configurations.

Allows routing different memory layers to different backends. For example: episodic -> MongoDB, semantic -> Postgres, working -> in-memory.

Any layer not listed in the override dicts falls back to the defaults supplied to :meth:OrchestratorFactory.build_hybrid.

OrchestratorFactory

Builds a fully-wired MemoryOrchestrator from configuration and backends.

build staticmethod

build(
    config: OrchestratorConfig,
    graph_backend: GraphBackend,
    vector_backend: VectorBackend,
    embedding_provider: EmbeddingProvider,
    llm: Any,
    catalog: ToolCatalog | None = None,
) -> MemoryOrchestrator

Wire all internal components from configuration and backends.

Parameters:

Name	Type	Description	Default
config	OrchestratorConfig	Orchestrator configuration.	required
graph_backend	GraphBackend	Graph storage backend.	required
vector_backend	VectorBackend	Vector storage backend.	required
embedding_provider	EmbeddingProvider	Text embedding provider.	required
llm	Any	LLM instance (BaseChatModel from langchain-core).	required
catalog	ToolCatalog | None	Optional tool catalog. Creates empty one if not provided.	None

Returns:

Type	Description
MemoryOrchestrator	Fully wired MemoryOrchestrator instance.

Source code in src/symfonic/memory/orchestrator/factory.py

@staticmethod
def build(
    config: OrchestratorConfig,
    graph_backend: GraphBackend,
    vector_backend: VectorBackend,
    embedding_provider: EmbeddingProvider,
    llm: Any,
    catalog: ToolCatalog | None = None,
) -> MemoryOrchestrator:
    """Wire all internal components from configuration and backends.

    Args:
        config: Orchestrator configuration.
        graph_backend: Graph storage backend.
        vector_backend: Vector storage backend.
        embedding_provider: Text embedding provider.
        llm: LLM instance (BaseChatModel from langchain-core).
        catalog: Optional tool catalog. Creates empty one if not provided.

    Returns:
        Fully wired MemoryOrchestrator instance.
    """
    from symfonic.memory.orchestrator.orchestrator import MemoryOrchestrator

    embedding_cache = EmbeddingCache() if embedding_provider is not None else None
    graph_store = GraphMemoryStore(
        graph_backend,
        embedding_provider=embedding_provider,
        embedding_cache=embedding_cache,
    )
    traversal = GraphTraversal(graph_store)
    scorer = RetrievalScorer(
        config.scoring_weights, scope_blend=config.scope_blend,
    )

    retrieval = RetrievalEngine(
        graph_store=graph_store,
        vector_backend=vector_backend,
        scorer=scorer,
        traversal=traversal,
        config=config,
    )

    layers = _build_layers(
        config, graph_store, vector_backend, embedding_provider, embedding_cache,
    )

    router = _build_router(config, layers, graph_store, llm, catalog)

    return MemoryOrchestrator(
        config=config,
        retrieval_engine=retrieval,
        router=router,
        layers=layers,
        embedding_provider=embedding_provider,
        embedding_cache=embedding_cache,
        graph_store=graph_store,
    )

build_hybrid staticmethod

build_hybrid(
    config: OrchestratorConfig,
    default_graph: GraphBackend,
    default_vector: VectorBackend,
    default_embedding: EmbeddingProvider,
    hybrid: HybridBackendConfig,
    llm: Any,
    catalog: ToolCatalog | None = None,
) -> MemoryOrchestrator

Build an orchestrator with per-layer backend routing.

For each enabled layer, checks hybrid for an override backend. Falls back to the supplied defaults for layers without overrides.

Parameters:

Name	Type	Description	Default
config	OrchestratorConfig	Orchestrator configuration.	required
default_graph	GraphBackend	Default graph backend for layers without an override.	required
default_vector	VectorBackend	Default vector backend for layers without an override.	required
default_embedding	EmbeddingProvider	Default embedding provider for layers without an override.	required
hybrid	HybridBackendConfig	Per-layer backend override map.	required
llm	Any	LLM instance (BaseChatModel from langchain-core).	required
catalog	ToolCatalog | None	Optional tool catalog.	None

Returns:

Type	Description
MemoryOrchestrator	Fully wired MemoryOrchestrator using per-layer routing.

Source code in src/symfonic/memory/orchestrator/factory.py

@staticmethod
def build_hybrid(
    config: OrchestratorConfig,
    default_graph: GraphBackend,
    default_vector: VectorBackend,
    default_embedding: EmbeddingProvider,
    hybrid: HybridBackendConfig,
    llm: Any,
    catalog: ToolCatalog | None = None,
) -> MemoryOrchestrator:
    """Build an orchestrator with per-layer backend routing.

    For each enabled layer, checks ``hybrid`` for an override backend.
    Falls back to the supplied defaults for layers without overrides.

    Args:
        config: Orchestrator configuration.
        default_graph: Default graph backend for layers without an override.
        default_vector: Default vector backend for layers without an override.
        default_embedding: Default embedding provider for layers without an override.
        hybrid: Per-layer backend override map.
        llm: LLM instance (BaseChatModel from langchain-core).
        catalog: Optional tool catalog.

    Returns:
        Fully wired MemoryOrchestrator using per-layer routing.
    """
    from symfonic.memory.orchestrator.orchestrator import MemoryOrchestrator

    embedding_cache = EmbeddingCache() if default_embedding is not None else None
    default_graph_store = GraphMemoryStore(
        default_graph,
        embedding_provider=default_embedding,
        embedding_cache=embedding_cache,
    )
    traversal = GraphTraversal(default_graph_store)
    scorer = RetrievalScorer(
        config.scoring_weights, scope_blend=config.scope_blend,
    )

    retrieval = RetrievalEngine(
        graph_store=default_graph_store,
        vector_backend=default_vector,
        scorer=scorer,
        traversal=traversal,
        config=config,
    )

    layers = _build_layers_hybrid(
        config=config,
        default_graph=default_graph,
        default_vector=default_vector,
        default_embedding=default_embedding,
        hybrid=hybrid,
        embedding_cache=embedding_cache,
    )

    router = _build_router(config, layers, default_graph_store, llm, catalog)

    return MemoryOrchestrator(
        config=config,
        retrieval_engine=retrieval,
        router=router,
        layers=layers,
        embedding_provider=default_embedding,
        embedding_cache=embedding_cache,
        graph_store=default_graph_store,
    )

from_postgres staticmethod

from_postgres(
    config: OrchestratorConfig,
    dsn: str,
    embedding_provider: EmbeddingProvider,
    llm: Any,
    catalog: ToolCatalog | None = None,
    min_pool_size: int = 2,
    max_pool_size: int = 10,
    embedding_dim: int = 1536,
) -> MemoryOrchestrator

Build a fully-wired orchestrator backed by PostgreSQL.

Creates a shared PostgresPoolManager and wires both graph and vector backends from it. Callers are responsible for opening the pool before use and closing it on shutdown.

Parameters:

Name	Type	Description	Default
config	OrchestratorConfig	Orchestrator configuration.	required
dsn	str	asyncpg-compatible DSN (e.g. postgresql://user:pw@host/db).	required
embedding_provider	EmbeddingProvider	Text embedding provider.	required
llm	Any	LLM instance (BaseChatModel from langchain-core).	required
catalog	ToolCatalog | None	Optional tool catalog.	None
min_pool_size	int	Minimum pool connections (default 2).	2
max_pool_size	int	Maximum pool connections (default 10).	10
embedding_dim	int	Vector embedding dimensions (default 1536).	1536

Returns:

Type	Description
MemoryOrchestrator	Fully wired MemoryOrchestrator using PostgreSQL backends.

Source code in src/symfonic/memory/orchestrator/factory.py

@staticmethod
def from_postgres(
    config: OrchestratorConfig,
    dsn: str,
    embedding_provider: EmbeddingProvider,
    llm: Any,
    catalog: ToolCatalog | None = None,
    min_pool_size: int = 2,
    max_pool_size: int = 10,
    embedding_dim: int = 1536,
) -> MemoryOrchestrator:
    """Build a fully-wired orchestrator backed by PostgreSQL.

    Creates a shared PostgresPoolManager and wires both graph and vector
    backends from it. Callers are responsible for opening the pool before
    use and closing it on shutdown.

    Args:
        config: Orchestrator configuration.
        dsn: asyncpg-compatible DSN (e.g. ``postgresql://user:pw@host/db``).
        embedding_provider: Text embedding provider.
        llm: LLM instance (BaseChatModel from langchain-core).
        catalog: Optional tool catalog.
        min_pool_size: Minimum pool connections (default 2).
        max_pool_size: Maximum pool connections (default 10).
        embedding_dim: Vector embedding dimensions (default 1536).

    Returns:
        Fully wired MemoryOrchestrator using PostgreSQL backends.
    """
    from symfonic.memory.backends.pool import PostgresPoolManager
    from symfonic.memory.backends.postgres_graph import PostgresGraphBackend
    from symfonic.memory.backends.postgres_vector import PostgresVectorBackend

    pool = PostgresPoolManager(
        dsn=dsn, min_size=min_pool_size, max_size=max_pool_size
    )
    graph_backend = PostgresGraphBackend(pool)
    vector_backend = PostgresVectorBackend(pool, embedding_dim=embedding_dim)

    return OrchestratorFactory.build(
        config=config,
        graph_backend=graph_backend,
        vector_backend=vector_backend,
        embedding_provider=embedding_provider,
        llm=llm,
        catalog=catalog,
    )