Integration Guide¶

Step-by-step guide for integrating symfonic-core into your application, from a minimal agent to a full-featured deployment with persistent memory, FastAPI endpoints, and domain plugins.

Progressive Guides¶

For a step-by-step learning path from basic to production:

Quickstart -- Your first agent in 10 lines
Memory Persistence -- Postgres + HMS
Streaming & Callbacks -- Real-time events
Plugins & Domains -- Custom domain logic
Production -- Full deployment topology
Lazy Tooling -- Reduce LLM token cost with per-query tool filtering

Installation¶

# In-memory only (no external services)
pip install symfonic-core

# Postgres graph + vector backends
pip install symfonic-core[postgres]

# MongoDB secondary (for graph visualization)
pip install symfonic-core[mongodb]

# Full stack
pip install symfonic-core[postgres,mongodb]

# FastAPI HTTP bridge
pip install symfonic-core[agent-api]

# Real LLM (Anthropic Claude)
pip install symfonic-core[anthropic]

Minimal Agent (10 Lines)¶

The simplest way to run a SymfonicAgent. No external services needed.

import asyncio
from symfonic.agent.config import FrameworkConfig
from symfonic.agent.engine import SymfonicAgent
from symfonic.agent.types import FrameworkTenantScope
from symfonic.core.testing import MockModelProvider

async def main():
    agent = SymfonicAgent(
        model_provider=MockModelProvider(response="Hello!"),
        config=FrameworkConfig(auto_hydrate=False, auto_consolidate=False),
    )
    scope = FrameworkTenantScope(tenant_id="my-app")
    response = await agent.run("Hello", scope=scope)
    print(response.final_response)  # "Hello!"
    print(f"Duration: {response.duration_ms:.1f}ms")

asyncio.run(main())

To use a real LLM, replace the mock provider:

from symfonic.core.providers import AnthropicProvider

agent = SymfonicAgent(
    model_provider=AnthropicProvider(),  # uses ANTHROPIC_API_KEY env var
    config=FrameworkConfig(auto_hydrate=False, auto_consolidate=False),
)

Postgres-Backed Agent with Memory¶

Enable the full 5-layer Hierarchical Memory System (HMS) with Postgres persistence. Memory survives restarts and is isolated per tenant.

Prerequisites¶

Start Postgres with pgvector:

docker compose up -d  # or provide your own Postgres with pgvector
export POSTGRES_DSN="postgresql://symfonic:symfonic_dev@localhost:5432/symfonic"

Code¶

import asyncio
import os
from symfonic.agent.config import FrameworkConfig
from symfonic.agent.engine import SymfonicAgent
from symfonic.agent.types import FrameworkTenantScope
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


class SimpleEmbeddingProvider:
    """Minimal embedding provider. Replace with a real one for production."""
    @property
    def dimensions(self) -> int:
        return 384

    async def embed(self, texts: list[str]) -> list[list[float]]:
        # Use sentence-transformers, OpenAI, etc. in production
        results = []
        for text in texts:
            h = hash(text) & 0xFFFFFFFF
            vec = [((h >> (i % 32)) & 1) * 0.1 for i in range(384)]
            results.append(vec)
        return results


async def main():
    dsn = os.environ["POSTGRES_DSN"]
    pool = PostgresPoolManager(dsn=dsn, min_size=2, max_size=10)
    await pool.open()

    graph_backend = PostgresGraphBackend(pool)
    vector_backend = PostgresVectorBackend(pool)
    await graph_backend.ensure_schema()
    await vector_backend.ensure_schema()

    agent = SymfonicAgent(
        model_provider=AnthropicProvider(),
        embedding_provider=SimpleEmbeddingProvider(),
        graph_backend=graph_backend,
        vector_backend=vector_backend,
        config=FrameworkConfig(
            auto_hydrate=True,       # inject memory context before LLM
            auto_consolidate=True,   # extract + commit memories after LLM
            spreading_activation=True,
            enable_hms_prompt=True,  # inject HMS system prompt sections
        ),
    )

    scope = FrameworkTenantScope(tenant_id="acme-corp")

    # First conversation -- agent learns
    r1 = await agent.run("Our brand voice is friendly and expert.", scope=scope)
    print(r1.final_response)

    # Second conversation -- agent remembers
    r2 = await agent.run("What do you know about our brand?", scope=scope)
    print(r2.final_response)
    print(f"Memory entries used: {r2.memory_entries_used}")

    await pool.close()

asyncio.run(main())

Key parameters:

enable_hms_prompt=True -- Injects the HMS system prompt section that instructs the LLM to emit <GRAPH_OPERATIONS> blocks for memory extraction.
auto_hydrate=True -- Before each LLM call, reads relevant memory from all 5 layers and injects it into the system prompt.
auto_consolidate=True -- After each LLM call, parses extraction blocks and commits new facts to the semantic graph.
spreading_activation=True -- When semantic retrieval produces hits, automatically fetches 1st-degree neighbor nodes for richer context.

Full FastAPI Application¶

Wrap a SymfonicAgent in HTTP endpoints with a single function call.

from fastapi import FastAPI
from symfonic.agent.config import FrameworkConfig
from symfonic.agent.engine import SymfonicAgent
from symfonic.agent.fastapi.router import create_agent_router
from symfonic.core.providers import AnthropicProvider

# 1. Build the agent
agent = SymfonicAgent(
    model_provider=AnthropicProvider(),
    config=FrameworkConfig(auto_hydrate=False, auto_consolidate=False),
)

# 2. Create the FastAPI app
app = FastAPI(title="My Agent API", version="1.0.0")

# 3. Mount the agent router
router = create_agent_router(agent, prefix="/api/v1")
app.include_router(router)

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

Run with:

uvicorn my_app:app --reload

Multi-Tenant Headers¶

All agent endpoints extract tenant context from HTTP headers:

Header	Required	Description
`X-Tenant-ID`	Yes	Unique tenant identifier
`X-Sub-Tenant-ID`	No	Sub-tenant identifier
`X-Namespace`	No	Memory namespace

Example request:

curl -X POST http://localhost:8000/api/v1/chat \
  -H "Content-Type: application/json" \
  -H "X-Tenant-ID: acme-corp" \
  -d '{"query": "Hello!", "tenant_id": "acme-corp"}'

Adding Domain Plugins¶

Domain plugins inject business-specific system prompts and guardrails into the agent without modifying the framework.

Plugin Structure¶

A domain plugin is any class with these methods:

class MyDomainPlugin:
    name = "My Domain"

    async def inject_system_prompt(self, state: dict) -> str:
        """Return domain-specific system prompt text."""
        return "You are a specialist in ..."

    async def validate_state_transition(
        self, action: str, context: dict
    ) -> bool:
        """Return False to block unsafe actions."""
        blocked = {"delete_account", "transfer_funds"}
        return action not in blocked

    def get_domain_tools(self) -> list:
        """Return LangChain tools for this domain.

        IMPORTANT: Tools must be passed at agent construction time,
        not via this method. Return [] here and pass tools via
        SymfonicAgent(tools=[...]).
        """
        return []

Loading a Plugin¶

from myapp.plugins import StoreGrowthPlugin

# Tools must be provided at construction time
plugin = StoreGrowthPlugin(monthly_fixed_costs=500.0)
domain_tools = plugin.get_domain_tools()

agent = SymfonicAgent(
    model_provider=provider,
    tools=domain_tools,        # tools go here
    config=FrameworkConfig(enable_hms_prompt=True),
)
agent.load_plugin(plugin)      # prompt + guardrails loaded here

Validating Actions¶

allowed = await agent.validate_action("delete_listing", {"tenant": "acme"})
if not allowed:
    print("Action blocked by domain guardrails")

Graph Queries with label_prefix¶

Since v5.4.0, GraphMemoryStore.query_nodes() supports label_prefix for prefix matching on node labels. This is essential when labels follow the "PREFIX: detail" convention (e.g. "SOUL: Joy", "SOUL: name: Amiel").

from symfonic.memory.graph.store import GraphMemoryStore
from symfonic.memory.types import MemoryLayer

# Exact label match -- only returns nodes labeled exactly "SOUL"
exact = await graph_store.query_nodes(scope, label="SOUL")

# Prefix match -- returns "SOUL: Joy", "SOUL: name: Amiel", etc.
prefixed = await graph_store.query_nodes(scope, label_prefix="SOUL")

# Combine with layer filter
semantic_soul = await graph_store.query_nodes(
    scope,
    layer=MemoryLayer.SEMANTIC,
    label_prefix="SOUL",
)

# label and label_prefix can coexist (AND logic)
specific = await graph_store.query_nodes(
    scope,
    label="SOUL: Joy",
    label_prefix="SOUL",
)

All three backends implement prefix matching safely:

Backend	Implementation
Postgres	`LIKE $N \|\| '%'` with positional parameter (injection-safe)
InMemory	`str.startswith()`
MongoDB	`re.escape()` anchored regex `^PREFIX`

Configuration Reference¶

FrameworkConfig is the top-level configuration object. All fields are optional with sensible defaults.

from symfonic.agent.config import FrameworkConfig

config = FrameworkConfig(
    auto_hydrate=True,          # Hydrate memory before LLM execution
    auto_consolidate=True,      # Extract and commit memories after execution
    spreading_activation=True,  # Fetch neighbor nodes on semantic hits
    self_reflection=False,      # Post-interaction insight extraction
    streaming_enabled=True,     # Enable the stream() interface
    enabled_layers={            # Which memory layers to hydrate
        "working",
        "episodic",
        "semantic",
        "procedural",
        "prospective",
    },
    reflection_prompt=None,     # Custom extraction prompt (must have {conversation})
)

FrameworkConfig Fields¶

Field	Type	Default	Description
`auto_hydrate`	`bool`	`True`	Inject memory context into system prompt before LLM
`auto_consolidate`	`bool`	`True`	Parse and commit new memories after LLM response
`spreading_activation`	`bool`	`True`	Fetch 1st-degree neighbor nodes on semantic hits
`self_reflection`	`bool`	`False`	Run InsightExtractor after each interaction
`streaming_enabled`	`bool`	`True`	Enable SSE streaming via `agent.stream()`
`enabled_layers`	`set[str]`	All 5	Memory layers to activate during hydration
`lazy_tooling`	`bool`	`True`	Dynamically filter tools via CapabilityRouter
`scheduler`	`Any`	`None`	TaskSchedulerProtocol for background tasks
`reflection_prompt`	`str`	`None`	Custom prompt for InsightExtractor

DomainTemplate Fields¶

Field	Type	Default	Description
`name`	`str`	`"default"`	Human-readable domain identifier
`required_labels`	`list[str]`	`[]`	Graph labels the domain expects (e.g. SOUL, PRODUCT)
`onboarding_checklist`	`list[str]`	`[]`	Items to collect during initial conversations
`soul_schema`	`dict[str, str]`	`{}`	Schema for the SOUL label (field name to type)
`tool_manifest`	`list[str]`	`[]`	Short tool descriptions injected into system prompt
`tool_catalog`	`dict`	`{}`	Full JSON schemas for JIT tool hydration
`monitoring`	`MonitoringBlueprint`	`(empty)`	Anomaly detection thresholds and intervals

Example: Custom Domain¶

from symfonic.agent.domain import DomainTemplate, MonitoringBlueprint

my_domain = DomainTemplate(
    name="healthcare-assistant",
    required_labels=["SOUL", "PATIENT_PROFILE", "TREATMENT_PLAN"],
    onboarding_checklist=[
        "Patient Name",
        "Primary Condition",
        "Current Medications",
    ],
    soul_schema={
        "specialty": "str",
        "communication_style": "str",
        "risk_tolerance": "str",
    },
    tool_manifest=[
        "lookup_drug: Check drug interactions and dosage guidelines",
        "schedule_appointment: Book a follow-up appointment",
    ],
    monitoring=MonitoringBlueprint(
        anomaly_thresholds={"vitals": 15.0},
        check_interval_seconds=600,
        alert_on=["blood_pressure", "heart_rate"],
    ),
)

config = FrameworkConfig(domain=my_domain)
agent = SymfonicAgent(model_provider=provider, config=config)

Memory Layers¶

The HMS provides 5 memory layers, each serving a distinct cognitive function:

Layer	Purpose	Backend
Semantic	Long-term knowledge graph (facts, entities, relationships)	Graph
Episodic	Past interaction records (vector-indexed for similarity search)	Vector
Working	Short-term context for the current conversation	Graph
Procedural	Learned workflows and skills (step-by-step procedures)	Graph
Prospective	Future goals and intentions	Graph

Enable or disable layers via FrameworkConfig.enabled_layers:

config = FrameworkConfig(
    enabled_layers={"semantic", "episodic", "working"},  # skip procedural + prospective
)

Logging & Observability¶

CallbackHandler -- Capturing LLM Prompts¶

Use the callbacks= parameter on run() or stream() to attach handlers that capture LLM-level events. This is the recommended way to log prompts and responses for debugging or compliance.

from symfonic.core.callbacks.protocol import CallbackHandler

class PromptAuditor(CallbackHandler):
    """Logs every LLM prompt and response for audit trail."""

    async def on_llm_start(self, event) -> None:
        print(f"[LLM START] model={event.model}")
        print(f"  messages={event.messages[:200]}...")

    async def on_llm_end(self, event) -> None:
        print(f"[LLM END] tokens={event.usage}")
        print(f"  response={event.response[:200]}...")

# Attach to a single call
response = await agent.run(
    "Summarise our sales",
    scope=scope,
    callbacks=[PromptAuditor()],
)

You can attach multiple handlers -- they run in parallel via CallbackManager. The handler protocol also includes on_agent_start, on_agent_end, on_node_start, on_node_end, and on_node_error for full lifecycle observability.

FrameworkObservabilityHook¶

For framework-level events (hydration, consolidation, routing, token budget), implement FrameworkObservabilityHook:

from symfonic.agent.hooks import LoggingFrameworkHook

agent = SymfonicAgent(
    model_provider=provider,
    config=FrameworkConfig(observability_hook=LoggingFrameworkHook()),
)

Next Steps¶

API Reference -- Full endpoint documentation for create_agent_router()
Examples -- Catalog of all example applications
Architecture -- Module map and design decisions
Memory Concepts -- Deep dive into the 5-layer HMS