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.

Legacy 9.12 example: Install symfonic-core==9.12.* for this SymfonicAgent snippet. For version 11, see the Agent migration guide.

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:

Legacy 9.12 example: Install symfonic-core==9.12.* for this SymfonicAgent snippet. For version 11, see the Agent migration guide.

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¶

Legacy 9.12 example: Install symfonic-core==9.12.* for this SymfonicAgent snippet. For version 11, see the Agent migration guide.

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.

Legacy 9.12 example: Install symfonic-core==9.12.* for this SymfonicAgent snippet. For version 11, see the Agent migration guide.

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:

Legacy 9.12 example: Install symfonic-core==9.12.* for this SymfonicAgent snippet. For version 11, see the Agent migration guide.

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¶

Legacy 9.12 example: Install symfonic-core==9.12.* for this SymfonicAgent snippet. For version 11, see the Agent migration guide.

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¶

Legacy 9.12 example: Install symfonic-core==9.12.* for this SymfonicAgent snippet. For version 11, see the Agent migration guide.

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:

Legacy 9.12 example: Install symfonic-core==9.12.* for this SymfonicAgent snippet. For version 11, see the Agent migration guide.

from symfonic.agent.observability 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