symfonic.agent -- Orchestration Layer¶

symfonic.agent is the top layer of the symfonic stack. It wires symfonic.core (execution engine) and symfonic.memory (5-layer HMS) into a single entry point: SymfonicAgent.

Dependency Chain¶

symfonic.agent
  -- imports -->  symfonic.core   (AgentRuntime, BaseAgentDeps, AgentGraph)
  -- imports -->  symfonic.memory (MemoryOrchestrator, layers, TenantScope)

Neither symfonic.core nor symfonic.memory import from symfonic.agent.

SymfonicAgent¶

SymfonicAgent is the main class. It owns one AgentRuntime and one MemoryOrchestrator. On every run() or stream() call it:

Hydrates -- queries all enabled non-working memory layers, runs spreading activation to expand hits with graph neighbors, and injects the assembled context into the LLM state.
Executes -- delegates to AgentRuntime.run() or streams via EventTranspiler for typed event output.
Extracts -- parses structured MEMORY_EXTRACT / GRAPH_OPERATIONS blocks from the LLM response using a 4-strategy parser.
Consolidates -- writes extracted ops to the appropriate memory layers, mirrors semantic nodes to MongoDB (optional), writes thinking nodes to working memory, and records an episodic turn summary.

Constructor Parameters¶

Parameter	Type	Default	Description
`model_provider`	`ModelProvider`	required	LLM provider (Anthropic, OpenAI, Mock)
`graph_preset`	`str`	`"react_loop"`	Graph topology preset
`config`	`FrameworkConfig`	`FrameworkConfig.with_defaults()`	Full framework config
`graph_backend`	`GraphBackend`	None	Custom graph backend (HMS)
`vector_backend`	`VectorBackend`	None	Custom vector backend (HMS)
`embedding_provider`	`EmbeddingProvider`	None	Embedding provider (HMS)
`tools`	`list`	None	Extra tools added to the graph
`orchestrator`	`MemoryOrchestrator`	None	Provide a pre-built orchestrator
`enable_hms_prompt`	`bool`	`False`	Inject HMS system prompt sections

If embedding_provider is given but backends are not, SymfonicAgent uses InMemory* backends automatically. If neither embedding_provider nor orchestrator is given, an empty orchestrator is created (all layers disabled).

run()¶

response: AgentResponse = await agent.run(
    "Summarise our last meeting",
    scope=FrameworkTenantScope(tenant_id="acme", sub_tenant_id="engineering"),
    callbacks=[my_callback_handler],
    session_id="sess-42",
    history=[prior_message_1, prior_message_2],
)

AgentResponse¶

AgentResponse is a frozen Pydantic model returned by run():

Field	Type	Description
`final_response`	`str \\| None`	LLM response text (extraction blocks stripped)
`messages`	`list[dict]`	Full message history
`memory_entries_used`	`int`	Entries retrieved during hydration
`system_prompt_tokens`	`int`	Token count of the system prompt
`run_id`	`str`	12-character hex run identifier
`duration_ms`	`float`	Wall-clock execution time
`extracted_ops`	`list[dict]`	Parsed memory operations from the LLM
`graph_edges`	`list[dict]`	Edges discovered via spreading activation
`activation_log`	`dict`	Spreading activation metadata (see below)

activation_log Schema¶

{
  "activated_nodes": [
    {"name": "Marinos Restaurant", "score": 0.85},
    {"name": "Ribeye Steak", "score": 0.72}
  ],
  "inference_paths": [
    ["Marinos Restaurant", "Ribeye Steak"],
    ["Marinos Restaurant", "Pescado Zarandeado"]
  ],
  "pending_connections": [
    {
      "source": "Marinos Restaurant",
      "relationship": "ASSOCIATED",
      "target": "Downtown Location",
      "status": "PENDING CONSOLIDATION"
    }
  ]
}

activated_nodes: All nodes that fired during spreading activation (seed hits + neighbors), deduplicated by name, sorted by score descending.
inference_paths: Pairs of [source_label, target_label] showing how each neighbor was reached from a seed node.
pending_connections: Inferred edges from generic relationships (associated, linked_semantic, linked_episodic) that need consolidation to become real graph edges.

Streaming¶

stream()¶

async for chunk in agent.stream("Tell me a story", scope=scope):
    print(chunk.event_type, chunk.data)

stream() yields StreamChunk objects. The standard event sequence is:

thinking -> [spreading_activation] -> text_delta (repeated) -> consolidating -> done

stream_text()¶

The simplest streaming API. Returns an async generator that yields only clean text strings -- no event wrappers, no extraction tags, no metadata.

async for text in agent.stream_text("Tell me a story", scope=scope):
    print(text, end="", flush=True)

Internally stream_text() pipes through ExtractionFilter so consumers never see <MEMORY_EXTRACT> or <GRAPH_OPERATIONS> blocks.

ExtractionFilter¶

ExtractionFilter is a stateful filter that strips <MEMORY_EXTRACT> and <GRAPH_OPERATIONS> blocks from streaming text. It handles the case where an XML tag is split across two or more chunks by buffering text when a < prefix is detected at the tail of a chunk. Call flush() at the end of the stream to drain any buffered non-tag remainder.

stream_typed() also uses ExtractionFilter internally -- TextDeltaEvent.text is always clean.

Lifecycle Events¶

Three lifecycle StreamEvent types are emitted during streaming:

Event	When emitted	Data
`MemoryExtractedEvent`	After extraction blocks are parsed from the response	Extracted operations
`ConsolidationDoneEvent`	After background memory write completes	Consolidation summary
`ResponseCompleteEvent`	After all processing finishes	`AgentResponse` summary

These events fire after the text stream completes, giving consumers hooks for post-response workflows (logging, analytics, UI notifications).

StreamChunk¶

Field	Type	Description
`event_type`	`EventType`	Event category
`data`	`Any`	Event payload
`timestamp`	`datetime`	UTC timestamp
`run_id`	`str`	12-character hex run identifier

Event Types¶

Event	When emitted	Data
`thinking`	Before any I/O starts	None
`spreading_activation`	After hydration, if activation data exists	`activation_log` dict
`text_delta`	Each LLM text chunk	Text string
`acting`	Agent is processing	Status message
`tool_call`	LLM invokes a tool	Tool name and args
`tool_result`	Tool returns a result	Tool output
`consolidating`	Background memory write starts	None
`done`	Stream complete	Summary dict with `run_id`, `duration_ms`, etc.

EventTranspiler¶

The EventTranspiler (in symfonic.core.streaming.transpiler) converts raw LangGraph astream_events(v2) into typed TimestampedEvent objects with millisecond precision. It handles:

Thinking blocks: Anthropic extended thinking content blocks with type="thinking" are emitted as ThinkingDeltaEvent.
Text blocks: Standard text content blocks are emitted as TextDeltaEvent.
Tool calls: ToolCallStartEvent, ToolCallDeltaEvent, ToolResultEvent.
Usage tracking: UsageEvent with input_tokens and output_tokens.
Unknown events: Wrapped as ExtensionEvent (never dropped).

LLM Callbacks¶

The CallbackHandler protocol includes on_llm_start() and on_llm_end() methods, allowing consumers to capture exact prompts and responses at the LLM level via the public callbacks= API.

from symfonic.core.callbacks.protocol import CallbackHandler

class PromptLogger(CallbackHandler):
    async def on_llm_start(self, event: LLMStartEvent) -> None:
        print(f"Prompt ({event.model}): {event.messages[:100]}...")

    async def on_llm_end(self, event: LLMEndEvent) -> None:
        print(f"Response: {event.response[:100]}...")

response = await agent.run(
    "Hello",
    scope=scope,
    callbacks=[PromptLogger()],
)

The react node fires both ObservabilityHook AND CallbackManager for LLM events, so LLM-level observability works whether you use the framework hook protocol or the public callback API.

HMS System Prompt¶

When enable_hms_prompt=True, the system prompt includes 9 sections defined in prompts/templates/hms_system.txt:

Data Sovereignty -- Tenant and namespace scoping rules.
Memory Architecture -- The Pentad -- Description of 5 layers and their operations (upsert_node, create_event, register_skill, set_trigger).
Available Tools -- Tool manifest injected at runtime.
Memory Context -- Hydrated entries from all layers.
Linked Context -- Spreading activation results.
Response Guidelines -- Output formatting rules.
Memory Extraction -- Instructions for generating MEMORY_EXTRACT blocks.
Graph Operations -- Schema for structured extraction JSON.
Domain Overrides -- Business-specific instructions from DomainTemplate.

Observability¶

FrameworkObservabilityHook is a protocol with four async methods:

Method	When called
`on_hydration_complete`	After memory hydration finishes
`on_consolidation_complete`	After background consolidation finishes
`on_routing_complete`	After lazy tool routing resolves
`on_token_budget_computed`	When token budget savings are measured

Two implementations: NoOpFrameworkHook (default) and LoggingFrameworkHook.

Domain Plugins¶

DomainTemplate¶

External applications inject business-specific behavior by creating a DomainTemplate and passing it to FrameworkConfig:

from symfonic.agent.domain import DomainTemplate

template = DomainTemplate(
    name="restaurant-expert",
    required_labels=["SOUL", "MENU", "HOURS", "BRAND"],
    onboarding_checklist=["business name", "cuisine type", "location"],
    soul_schema={
        "business_name": "str",
        "cuisine_type": "str",
        "founded_year": "int",
    },
)

Field	Description
`name`	Human-readable domain identifier
`required_labels`	Graph labels the domain expects (used by DiscoveryService to flag missing blocks)
`onboarding_checklist`	Items the agent should collect in initial conversations
`soul_schema`	Schema for the SOUL label -- maps field names to expected types
`monitoring`	`MonitoringBlueprint` with anomaly thresholds and check intervals

The framework stays generic; all business-specific labels, onboarding flows, and schema definitions live in the consuming application.

FrameworkConfig¶

FrameworkConfig is a frozen Pydantic model composing AgentConfig and OrchestratorConfig with framework behavior flags:

Flag	Default	Description
`auto_hydrate`	`True`	Inject memory context before LLM call
`auto_consolidate`	`True`	Write final response to memory
`lazy_tooling`	`True`	Route tools via procedural layer per query
`streaming_enabled`	`True`	Allow `stream()` interface
`spreading_activation`	`False`	Enable graph neighbor expansion
`enabled_layers`	All five	Which memory layers to hydrate

FrameworkTenantScope¶

Bridges symfonic.core.TenantScope (frozen dataclass: tenant_id, sub_tenant_id) and symfonic.memory.TenantScope (Pydantic: tenant_id, namespace). Provides to_core_scope(), to_memory_scope(), and from_core_scope() / from_memory_scope() class methods.

When sub_tenant_id and namespace are both provided they must be equal. When only one is provided, the other is derived automatically on conversion.

HMSFactory¶

Stateless builder that constructs a fully wired MemoryOrchestrator:

from symfonic.agent import HMSFactory

# In-memory (dev/test)
orchestrator = HMSFactory.build_in_memory(embedding_provider=my_embedder)

# Custom backends (production)
orchestrator = HMSFactory.build(
    graph_backend=postgres_graph,
    vector_backend=postgres_vector,
    embedding_provider=my_embedder,
)

FastAPI Integration¶

Install the agent-api extra:

pip install "symfonic-core[agent-api]"

from fastapi import FastAPI
from symfonic.agent import SymfonicAgent
from symfonic.agent.fastapi.router import create_agent_router

agent = SymfonicAgent(model_provider=my_provider)
app = FastAPI()
app.include_router(create_agent_router(agent, prefix="/api/v1"))

Tenant scope is extracted from request headers:

Header	Required	Maps to
`X-Tenant-ID`	yes	`tenant_id`
`X-Sub-Tenant-ID`	no	`sub_tenant_id`
`X-Namespace`	no	`namespace`

Public API Reference¶

Symbol	Description
`SymfonicAgent`	Main agent class (run/stream/stream_text)
`FrameworkConfig`	Unified config composing core + memory
`FrameworkTenantScope`	Tenant scope bridging core and memory
`HMSFactory`	Stateless builder for MemoryOrchestrator
`AgentRequest`	Pydantic model for inbound requests
`AgentResponse`	Pydantic model for agent responses
`StreamChunk`	Single streaming event
`ExtractionFilter`	Stateful filter stripping extraction blocks from text
`DomainTemplate`	Business-specific domain configuration