Level 5: Production Deployment

Deploy symfonic with offline consolidation, spreading activation, graph traversal, hybrid backends, and full observability. This level covers the complete production topology.

Prerequisites

• Completed Levels 1-4
• Docker and Docker Compose
• Familiarity with Celery (optional, for consolidation scheduling)

Architecture Overview

                          +-------------------+
                          |   FastAPI / SSE   |
                          |  create_agent_    |
                          |    router()       |
                          +--------+----------+
                                   |
                          +--------v----------+
                          |  SymfonicAgent    |
                          |  - run()          |
                          |  - stream_typed() |
                          |  - stream_text()  |
                          |  - load_plugin()  |
                          +--------+----------+
                                   |
              +--------------------+--------------------+
              |                    |                     |
    +---------v-------+  +--------v--------+  +---------v--------+
    | AgentRuntime    |  | MemoryOrch.     |  | Plugins          |
    | (LangGraph)     |  | (5-layer HMS)   |  | (prompt+guards)  |
    +---------+-------+  +--------+--------+  +------------------+
              |                    |
              |          +---------+---------+
              |          |                   |
         +----v----+  +--v---------+  +------v------+
         | LLM     |  | Postgres   |  | Vector      |
         | Provider|  | Graph      |  | (pgvector)  |
         +---------+  +-----+------+  +------+------+
                            |                |
                      +-----v----------------v-----+
                      |  PostgresPoolManager       |
                      |  (shared connection pool)  |
                      +----------------------------+

Docker Compose: Full Stack

# docker-compose.yml
version: "3.9"
services:
  postgres:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_USER: symfonic
      POSTGRES_PASSWORD: symfonic_dev
      POSTGRES_DB: symfonic
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  celery-worker:
    build: .
    command: celery -A my_app.celery worker --loglevel=info
    environment:
      POSTGRES_DSN: postgresql://symfonic:symfonic_dev@postgres:5432/symfonic
      REDIS_URL: redis://redis:6379/0
    depends_on:
      - postgres
      - redis

  celery-beat:
    build: .
    command: celery -A my_app.celery beat --loglevel=info
    environment:
      POSTGRES_DSN: postgresql://symfonic:symfonic_dev@postgres:5432/symfonic
      REDIS_URL: redis://redis:6379/0
    depends_on:
      - postgres
      - redis

  app:
    build: .
    command: uvicorn my_app:app --host 0.0.0.0 --port 8000
    ports:
      - "8000:8000"
    environment:
      POSTGRES_DSN: postgresql://symfonic:symfonic_dev@postgres:5432/symfonic
      REDIS_URL: redis://redis:6379/0
      ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY}
    depends_on:
      - postgres
      - redis

volumes:
  pgdata:

SleepConsolidator: Offline Memory Maintenance

SleepConsolidator runs 11 distinct phases of memory maintenance designed for nightly batch execution. Three entrypoints:

• quick_nap(scope) — lightweight subset (Phase 1 strengthen + optional Phase 5 SOUL + optional Phase 10 episodic). Fires every N turns when FrameworkConfig.quick_nap_interval > 0.
• run(scope) — full roster, returns a ConsolidationReport with phases_run: list[str] listing every phase that fired.
• nightly_nap(scope) — v7.0 T12 full-roster alias gated by FrameworkConfig.nightly_nap_enabled=True. Operators wire their own scheduler; engine does not parse the cron string.

Phase	Name	Effect
1	strengthen	Boost importance for recurring nodes (combines access_count + phase1_spreading_weight * spreading_access_count)
1.5	cooccurrence	Create CO_OCCURRED edges between co-temporal nodes
2	tag_risk	Mark nodes with negative-sentiment context
2.5	pending_edges	Write inferred pending_connections as real edges (conditional)
3	prune_orphans	Delete stale low-confidence orphan nodes; TTL-expired pruned unconditionally
4	meta_nodes	Cluster raw domain nodes by label prefix (excludes existing META: to prevent cascade)
5	soul_corrections	Apply SOUL schema updates (conditional on soul_schema= arg)
8	working_ttl	Prune WORKING-layer nodes with expired TTL
9	decay_importance	Reduce importance of stale nodes; skips importance >= 8.0 and SOUL: / AGENT_IDENTITY: labels
10	episodic_summary	Compress old episodes into semantic meta-nodes (gated by episodic_summarization_max_entries)
11	synthetic_links	Emit SYNTHETIC_LINK edges between semantic nodes that co-occur in episodic content (gated by synthetic_link_min_co_count)
12	procedural_promotion	Promote recurring episodic patterns to draft procedural skills (gated by promotion_min_pattern_count)

Running Consolidation Manually

import asyncio
from symfonic.core.learning.consolidation import SleepConsolidator
from symfonic.memory.graph.store import GraphMemoryStore
from symfonic.memory.types import TenantScope

async def consolidate(graph_store: GraphMemoryStore, tenant_id: str):
    consolidator = SleepConsolidator(
        graph_store=graph_store,
        lookback_hours=24,
    )
    scope = TenantScope(tenant_id=tenant_id)
    report = await consolidator.run(scope)

    print(f"Phases run: {report.phases_run}")
    print(f"Strengthened: {report.nodes_strengthened}")
    print(f"Pruned: {report.nodes_pruned}")
    print(f"Meta-nodes created: {report.meta_nodes_created}")
    print(f"SOUL updates: {report.soul_updates}")
    print(f"Working TTL pruned: {report.working_ttl_pruned}")
    print(f"Nodes decayed: {report.nodes_decayed}")
    print(f"Episodes summarized: {report.episodes_summarized}")
    print(f"Draft skills promoted: {report.draft_skills_promoted}")
    return report

report.phases_run lists every phase that actually executed (including zero-mutation phases). Use it to distinguish "phase ran and found nothing" from "phase was short-circuited".

Scheduling with Celery Beat

Use the CeleryTaskProvider adapter to schedule consolidation as a recurring background task:

from celery import Celery
from examples.full_demo.infra.celery_adapter import CeleryTaskProvider

celery_app = Celery("symfonic", broker="redis://localhost:6379/0")

# Register the consolidation task
@celery_app.task(name="execute_skill")
def execute_skill(skill_id: str, **kwargs):
    """Universal skill dispatcher."""
    if skill_id == "deep_sleep_consolidation":
        import asyncio
        from symfonic.core.learning.consolidation import SleepConsolidator

        tenant_id = kwargs.get("tenant_id", "default")
        # graph_store initialized from app context
        consolidator = SleepConsolidator(graph_store=graph_store, lookback_hours=24)
        scope = TenantScope(tenant_id=tenant_id)
        report = asyncio.run(consolidator.run(scope))
        return report.to_dict()

# Schedule via CeleryTaskProvider
async def setup_scheduling():
    provider = CeleryTaskProvider(celery_app)
    schedule_id = await provider.schedule_recurring(
        task_name="deep_sleep_consolidation",
        interval_seconds=86400,  # daily
        params={"tenant_id": "acme-corp"},
    )
    print(f"Scheduled: {schedule_id}")

The InProcessScheduler is available for single-process deployments without Celery:

from symfonic.agent.scheduler import InProcessScheduler

scheduler = InProcessScheduler()
config = FrameworkConfig(scheduler=scheduler)
agent = SymfonicAgent(model_provider=provider, config=config)

Spreading Activation and ActivationEvent

When spreading_activation=True, the agent automatically fetches 1st-degree neighbor nodes when semantic retrieval produces hits. This gives the LLM richer context without explicit queries.

Each activated node emits an ActivationEvent during streaming:

from symfonic.core.contracts.types import ActivationEvent

async for event in agent.stream_typed(query, scope=scope):
    if isinstance(event, ActivationEvent):
        print(
            f"Activated: {event.node_label} "
            f"(score={event.score:.2f}, layer={event.layer})"
        )
        if event.source_node_id:
            print(f"  triggered by: {event.source_node_id}")

ActivationEvent fields:

Field	Type	Description
node_id	str	Unique node identifier
node_label	str	Human-readable label (e.g. "SOUL: brand_voice")
score	float	Activation score 0.0 to 1.0
layer	str	Memory layer (semantic, working, episodic, etc.)
source_node_id	str | None	Parent node that triggered activation

Graph Traversal API

The GraphTraversal class provides BFS and shortest-path queries over the memory graph:

from symfonic.memory.graph.traversal import GraphTraversal
from symfonic.memory.types import TenantScope

graph_store = agent.orchestrator.graph_store
traversal = GraphTraversal(graph_store)
scope = TenantScope(tenant_id="acme-corp")

# BFS: find all nodes within 2 hops of a starting node
neighborhood = await traversal.bfs(scope, start="node-123", max_depth=2)
for node in neighborhood:
    print(f"  {node.label}: {node.content[:60]}")

# Shortest path between two nodes
path = await traversal.shortest_path(scope, source="node-A", target="node-B")
if path:
    print(f"Path: {' -> '.join(str(n) for n in path)}")
else:
    print("No path found")

Hybrid Backends: Per-Layer Storage

Route different memory layers to different backends using HybridBackendConfig. For example, episodic data to a vector-optimized store while semantic data stays in Postgres:

from symfonic.memory.orchestrator.factory import HybridBackendConfig, OrchestratorFactory
from symfonic.memory.types import MemoryLayer

hybrid_config = HybridBackendConfig(
    layer_graph_backends={
        MemoryLayer.SEMANTIC: postgres_graph,
        MemoryLayer.WORKING: in_memory_graph,
    },
    layer_vector_backends={
        MemoryLayer.EPISODIC: postgres_vector,
    },
)

# Build orchestrator with hybrid routing
orchestrator = OrchestratorFactory.build_hybrid(
    config=orchestrator_config,
    default_graph_backend=postgres_graph,
    default_vector_backend=postgres_vector,
    embedding_provider=embedding_provider,
    hybrid_config=hybrid_config,
)

agent = SymfonicAgent(
    model_provider=provider,
    orchestrator=orchestrator,
    config=FrameworkConfig(),
)

Multi-Tenant Isolation

Every operation in symfonic requires a TenantScope. This is enforced at the backend level -- there is no way to accidentally leak data between tenants.

from symfonic.agent.types import FrameworkTenantScope

# Full scope with sub-tenant and namespace
scope = FrameworkTenantScope(
    tenant_id="acme-corp",
    sub_tenant_id="west-coast",
    namespace="west-coast",       # must match sub_tenant_id if both set
)

HTTP headers for the FastAPI bridge:

Header	Required	Maps to
X-Tenant-ID	Yes	scope.tenant_id
X-Sub-Tenant-ID	No	scope.sub_tenant_id
X-Namespace	No	scope.namespace

Production Hardening: v6.1.9+ Security Knobs

Production auth fail-fast

create_agent_router() raises RuntimeError at build time when SYMFONIC_ENV=production (or APP_ENV / ENVIRONMENT / NODE_ENV set to production / prod) and no tenant-auth verifier was registered. Emergency override: ALLOW_INSECURE_PROD=true — logs CRITICAL and data will leak across tenants if left enabled.

from symfonic.agent.fastapi.auth import set_tenant_auth_verifier
from symfonic.agent.fastapi.router import create_agent_router

set_tenant_auth_verifier(my_jwt_verifier)  # MUST be called before router build
router = create_agent_router(agent, prefix="/api/v1")

Credential scrubbing (three-tier defence-in-depth)

Three layers scrub credential-shaped property keys on every memory op:

1. Extraction-time — HMS / JIT / extraction prompts instruct the LLM to omit credential-shaped properties.
2. Write-time — SymfonicAgent._consolidate scrubs before persist.
3. Read-time — SymfonicAgent._hydrate scrubs before formatting the context string.

Default pattern list: password, secret, token, api_key / api-key, credential, private_key / private-key, bearer, authorization, auth_header, access_key, refresh_token.

from symfonic.agent.config import FrameworkConfig

# Default: None = use built-in v6.1.9 list byte-for-byte
cfg = FrameworkConfig()

# Explicit opt-out (tenants that run their own upstream hygiene):
cfg = FrameworkConfig(credential_patterns=[])

# Custom list REPLACES built-in (invalid regex raises re.error):
cfg = FrameworkConfig(credential_patterns=[
    r"password", r"secret", r"auth_token", r"client_secret",
])

Public entrypoint for non-engine call sites: symfonic.agent.hygiene.scrub_credential_keys.

Intent filter + tool routing (v7.0 / v7.0.1)

For deployments with 10+ bound tools, narrow the per-turn tools= payload via tool_routing_mode. Safe rollout: start in observe mode to collect ToolRoutingDecisionEvent telemetry before flipping to enforce.

from symfonic.agent.config import FrameworkConfig
from symfonic.agent.domain import DomainTemplate

cfg = FrameworkConfig(
    intent_filter_mode="enforce",   # action turns skip semantic/episodic
    tool_routing_mode="enforce",    # narrow tools= payload per intent
    always_include_tools=("hydrate_context", "emit_telemetry"),
    domain=DomainTemplate(
        tool_trigger_keywords={
            "mysql_query": ["mysql", "query", "sql", "database"],
            "bash": ["bash", "shell", "command"],
        },
    ),
)

See Guide 07 for the full recipe.

Fabrication detection (v7.0 / v7.0.3)

In observe mode the detector emits telemetry but never mutates drafts — safe to enable in prod while tuning thresholds. Escalate to refuse once you trust the signal.

cfg = FrameworkConfig(
    fabrication_check_mode="refuse",
    fabrication_refuse_min_confidence=0.8,  # catches UUID (0.95) + kv/verb (0.8)
    # Lower to 0.5 to catch short-hex too; raise to 1.0 for observe-only
)

See Guide 08 for the deployment stance.

Self-Reflection: Insight Extraction

Enable post-interaction insight extraction to automatically create semantic graph nodes from conversations:

config = FrameworkConfig(
    self_reflection=True,
    reflection_prompt="""Analyze the conversation to extract insights.
For each insight, return a JSON array of objects with:
- "subject": entity name
- "relation": semantic link type
- "object": target entity or value
- "fact": 1-sentence summary
- "confidence": 0.0-1.0

Conversation:
{conversation}

Return ONLY a JSON array. If no insights found, return [].""",
)

The {conversation} placeholder is required. The InsightExtractor runs after each interaction and commits extracted facts to the semantic graph.

Monitoring and Observability

FrameworkObservabilityHook

from symfonic.agent.observability import FrameworkObservabilityHook

class MetricsCollector:
    """Collect framework metrics for Prometheus/Datadog/etc."""

    async def on_hydration_complete(self, tenant_id, query, entries_count, latency_ms):
        metrics.histogram("symfonic.hydration.latency", latency_ms, tags=[f"tenant:{tenant_id}"])
        metrics.gauge("symfonic.hydration.entries", entries_count, tags=[f"tenant:{tenant_id}"])

    async def on_consolidation_complete(self, tenant_id, operations_count, latency_ms):
        metrics.histogram("symfonic.consolidation.latency", latency_ms, tags=[f"tenant:{tenant_id}"])

    async def on_routing_complete(self, tenant_id, query, tools_selected, latency_ms):
        metrics.gauge("symfonic.routing.tools", len(tools_selected), tags=[f"tenant:{tenant_id}"])

    async def on_token_budget_computed(self, tenant_id, total_tool_tokens, filtered_tool_tokens, savings_pct):
        metrics.gauge("symfonic.tokens.savings_pct", savings_pct, tags=[f"tenant:{tenant_id}"])

CallbackHandler for Compliance Logging

class ComplianceLogger:
    """Log all LLM interactions for audit trail.

    v7.4.3: every non-react LLM call site (metacognition critic,
    context-window summariser, insight extractor, LLM entity extractor,
    consolidation extractor, procedural router) now fires
    ``on_llm_end`` with a populated ``node_name`` discriminator so the
    audit trail captures the full cost surface, not just the main
    reasoning loop.
    """

    async def on_llm_start(self, event):
        audit_log.write({
            "type": "llm_start",
            "model": event.model,
            "message_count": len(event.messages),
            "timestamp": datetime.utcnow().isoformat(),
        })

    async def on_llm_end(self, event):
        audit_log.write({
            "type": "llm_end",
            # v7.4.3: node_name attributes the call to react /
            # metacognition_critic / context_window_summariser / etc.
            # so compliance dashboards can slice cost by subsystem.
            "node_name": event.node_name,
            "model": event.model,
            "usage": event.usage,
            "timestamp": datetime.utcnow().isoformat(),
        })

    # Implement remaining hooks as no-ops
    async def on_agent_start(self, event): pass
    async def on_agent_end(self, event): pass
    async def on_node_start(self, event): pass
    async def on_node_end(self, event): pass
    async def on_node_error(self, event): pass

Performance Tips

Token Budget

Reduce system prompt size by limiting enabled layers and tool count:

config = FrameworkConfig(
    enabled_layers={"semantic", "episodic"},  # skip working/procedural/prospective
    lazy_tooling=True,                         # only inject relevant tools per query
)

Conversation History

Limit the number of prior messages injected:

from symfonic.memory.orchestrator.config import OrchestratorConfig

config = FrameworkConfig(
    orchestrator=OrchestratorConfig(
        context_budget=ContextBudget(max_history_messages=10),
    ),
)

Streaming Consolidation

Consolidation runs in a background task after streaming completes. Call flush_background_tasks() before shutdown to ensure all writes complete:

# In your FastAPI lifespan
@asynccontextmanager
async def lifespan(app):
    yield
    await agent.flush_background_tasks()

app = FastAPI(lifespan=lifespan)

Full Production Setup

Combining everything from Levels 1-5:

import asyncio
import os
from contextlib import asynccontextmanager

from fastapi import FastAPI

from symfonic.agent.config import FrameworkConfig
from symfonic.agent.domain import DomainTemplate
from symfonic.agent.engine import SymfonicAgent
from symfonic.agent.fastapi.router import create_agent_router
from symfonic.core.providers import AnthropicProvider
from symfonic.memory.backends.pool import PostgresPoolManager
from symfonic.memory.backends.postgres_graph import PostgresGraphBackend
from symfonic.memory.backends.postgres_vector import PostgresVectorBackend


MY_DOMAIN = DomainTemplate(
    name="my-product",
    required_labels=["SOUL", "USER_PROFILE"],
    soul_schema={"name": "str", "plan": "str"},
)

pool = None
agent = None

@asynccontextmanager
async def lifespan(app):
    global pool, agent
    dsn = os.environ["POSTGRES_DSN"]

    pool = PostgresPoolManager(dsn=dsn, min_size=2, max_size=20)
    await pool.open()

    graph = PostgresGraphBackend(pool)
    vector = PostgresVectorBackend(pool)
    await graph.ensure_schema()
    await vector.ensure_schema()

    agent = SymfonicAgent(
        model_provider=AnthropicProvider(),
        embedding_provider=my_embedding_provider,
        graph_backend=graph,
        vector_backend=vector,
        config=FrameworkConfig(
            domain=MY_DOMAIN,
            auto_hydrate=True,
            auto_consolidate=True,
            spreading_activation=True,
            self_reflection=True,
            enable_hms_prompt=True,
            # v7.0 production stance
            intent_filter_mode="observe",          # flip to enforce after burn-in
            tool_routing_mode="observe",
            fabrication_check_mode="observe",
            quick_nap_interval=10,
            nightly_nap_enabled=True,
        ),
    )

    router = create_agent_router(agent, prefix="/api/v1")
    app.include_router(router)

    yield

    await agent.flush_background_tasks()
    await pool.close()

app = FastAPI(title="Production Agent", lifespan=lifespan)

@app.get("/health")
async def health():
    return {"status": "ok"}

Run:

docker compose up -d
ANTHROPIC_API_KEY=sk-ant-... uvicorn my_app:app --host 0.0.0.0 --port 8000

Further Reading

• Feature Catalog -- All 11 capability clusters with opt-in examples
• Guide 07 Intent & Tool Routing -- v7.0.1 tool narrowing recipe
• Guide 08 Fabrication Detection -- Deployment stance and confidence thresholds
• Guide 09 Consolidation & Deep Sleep -- 11 phases in detail
• Integration Guide -- Configuration reference
• Memory Concepts -- Deep dive into 5-layer HMS
• Architecture -- Module map and Thalamic pipeline
• API Reference -- Full endpoint documentation