Level 3: Streaming and Observability

Stream agent responses token-by-token and observe every step of the execution pipeline. This level covers the three streaming APIs, the ExtractionFilter, and the callback system.

Prerequisites

• Completed Level 2
• A working agent with memory (Postgres or in-memory)

Three Streaming APIs

SymfonicAgent provides three streaming methods, each at a different abstraction level:

Method	Yields	Best for
stream_text()	str	Chat UI -- just the clean text
stream_typed()	StreamEvent subclasses	Rich UIs -- text, tool calls, thinking, activation
stream()	StreamChunk	Low-level -- raw event wrappers with event_type field

All three methods share the same signature:

async for chunk in agent.stream_text(
    query,
    scope=scope,
    callbacks=[...],       # optional per-call callbacks
    session_id="abc",      # optional session tracking
    history=[...],         # optional prior messages
):
    ...

stream_text() -- Simplest Streaming

For chat UIs that just need clean text, stream_text() yields only str values. All lifecycle events, tool calls, and extraction blocks are silently consumed (consolidation still runs).

import asyncio
from symfonic.agent.config import FrameworkConfig
from symfonic.agent.engine import SymfonicAgent
from symfonic.agent.types import FrameworkTenantScope
from symfonic.core.providers import AnthropicProvider

async def main():
    agent = SymfonicAgent(
        model_provider=AnthropicProvider(),
        config=FrameworkConfig(auto_hydrate=False, auto_consolidate=False),
    )
    scope = FrameworkTenantScope(tenant_id="demo")

    print("Agent: ", end="", flush=True)
    async for text in agent.stream_text("Explain dependency injection in 3 sentences.", scope=scope):
        print(text, end="", flush=True)
    print()

asyncio.run(main())

Output streams character-by-character as the LLM generates tokens.

stream_typed() -- Rich Event Stream

For UIs that show thinking indicators, tool call progress, or graph activation animations, stream_typed() yields typed event objects:

import asyncio
from symfonic.agent.config import FrameworkConfig
from symfonic.agent.engine import SymfonicAgent
from symfonic.agent.types import FrameworkTenantScope
from symfonic.core.contracts.types import (
    ActivationEvent,
    TextDeltaEvent,
    ThinkingDeltaEvent,
    ToolCallStartEvent,
    ToolResultEvent,
    UsageEvent,
)
from symfonic.core.providers import AnthropicProvider

async def main():
    agent = SymfonicAgent(
        model_provider=AnthropicProvider(),
        config=FrameworkConfig(auto_hydrate=False, auto_consolidate=False),
    )
    scope = FrameworkTenantScope(tenant_id="demo")

    async for event in agent.stream_typed("What tools do you have?", scope=scope):
        match event:
            case TextDeltaEvent(text=t):
                print(t, end="", flush=True)
            case ThinkingDeltaEvent(text=t):
                print(f"[thinking] {t}")
            case ToolCallStartEvent(tool_name=name):
                print(f"\n[tool] Calling {name}...")
            case ToolResultEvent(result=r):
                print(f"[tool] Result: {r[:100]}")
            case ActivationEvent(node_label=label, score=s):
                print(f"[memory] Activated: {label} (score={s:.2f})")
            case UsageEvent(input_tokens=inp, output_tokens=out):
                print(f"\n[usage] {inp} in / {out} out")
            case _:
                pass  # MessageStartEvent, MessageEndEvent, ExtensionEvent, etc.
    print()

asyncio.run(main())

The full event type hierarchy:

Event Type	Fields	When
TextDeltaEvent	text: str	Each LLM output token
ThinkingDeltaEvent	text: str	Extended thinking / chain-of-thought
ToolCallStartEvent	tool_name, tool_call_id	LLM decides to call a tool
ToolCallDeltaEvent	tool_call_id, args_delta	Incremental tool arguments
ToolResultEvent	tool_call_id, result	Tool execution completes
ActivationEvent	node_id, node_label, score, layer	Memory node activated via spreading activation
MessageStartEvent	message_id, model	New LLM message begins
MessageEndEvent	message_id, stop_reason	LLM message ends
UsageEvent	input_tokens, output_tokens	Token usage report
ExtensionEvent	type, payload	Vendor-specific or custom events

ExtractionFilter: Automatic Block Stripping

When enable_hms_prompt=True, the LLM emits <GRAPH_OPERATIONS> blocks inline with its response text. These blocks contain memory extraction instructions and must not be shown to the user.

The ExtractionFilter is a stateful streaming filter that handles this automatically. It works correctly even when tags are split across multiple chunks:

from symfonic.core.streaming.extraction_filter import ExtractionFilter

filt = ExtractionFilter()

# Simulate chunked LLM output with an embedded extraction block
chunks = [
    "Here is my answer. ",
    "<GRAPH_OPERA",           # partial tag -- buffered
    'TIONS>[{"type":"create_node","label":"SOUL"}]</GRAPH_OPERATIONS>',
    " Hope that helps!",
]

for chunk in chunks:
    clean = filt.feed(chunk)
    if clean:
        print(clean, end="")

remainder = filt.flush()
if remainder:
    print(remainder, end="")

print()
print(f"Extracted ops: {filt.get_extracted_ops()}")

Output:

Here is my answer.  Hope that helps!
Extracted ops: [{'type': 'create_node', 'label': 'SOUL'}]

Both stream_typed() and stream_text() apply the ExtractionFilter internally -- you never see extraction blocks in the output.

CallbackHandler: Prompt Auditing

Attach callbacks to any run() or stream() call for LLM-level observability. The CallbackHandler protocol defines 7 lifecycle hooks:

import asyncio
from symfonic.agent.config import FrameworkConfig
from symfonic.agent.engine import SymfonicAgent
from symfonic.agent.types import FrameworkTenantScope
from symfonic.core.callbacks.protocol import CallbackHandler
from symfonic.core.providers import AnthropicProvider


class PromptAuditor:
    """Logs every LLM prompt and response for compliance.

    As of v7.4.3, every LLM call site is wired -- this includes the
    metacognition critic, the always-on context-window summariser, the
    LLM extraction service, the procedural router, and more.  The
    ``LLMEndEvent.node_name`` discriminator identifies which subsystem
    issued the call so cost / latency can be attributed correctly:

    * ``"react"``                       -- main agent reasoning loop
    * ``"metacognition_critic"``        -- post-draft reflection pass
    * ``"metacognition_skill_gate"``    -- skill-promotion gating critic
    * ``"context_window_summariser"``   -- conversation compaction (always-on)
    * ``"insight_extractor"``           -- self-reflection insight extraction
    * ``"entity_extractor_llm"``        -- Phase 11 entity extraction
    * ``"consolidation_extractor"``     -- structured-ops extraction
    * ``"procedural_router"``           -- LLM-backed tool router
    """

    async def on_agent_start(self, event) -> None:
        print(f"[agent] Started run_id={event.run_id}")

    async def on_agent_end(self, event) -> None:
        print(f"[agent] Finished run_id={event.run_id}")

    async def on_node_start(self, event) -> None:
        print(f"  [node] -> {event.node_name}")

    async def on_node_end(self, event) -> None:
        print(f"  [node] <- {event.node_name} ({event.duration_ms:.0f}ms)")

    async def on_node_error(self, event) -> None:
        print(f"  [node] ERROR {event.node_name}: {event.error}")

    async def on_llm_start(self, event) -> None:
        print(f"  [llm] Sending {len(event.messages)} messages to {event.model}")

    async def on_llm_end(self, event) -> None:
        # v7.4.3: event.node_name discriminates between react/critic/compactor/etc.
        # Defaults to "" for handlers that don't care; non-empty for everything
        # emitted by symfonic's instrumented call sites.
        source = event.node_name or "(legacy)"
        print(f"  [llm:{source}] Response: {event.usage} tokens")


async def main():
    agent = SymfonicAgent(
        model_provider=AnthropicProvider(),
        config=FrameworkConfig(auto_hydrate=False, auto_consolidate=False),
    )
    scope = FrameworkTenantScope(tenant_id="demo")

    # Attach to a single call
    response = await agent.run(
        "Summarize the benefits of DI.",
        scope=scope,
        callbacks=[PromptAuditor()],
    )
    print(f"\nResponse: {response.final_response[:100]}...")

asyncio.run(main())

Multiple callbacks can be attached -- they run in parallel via CallbackManager.

Optional Hooks (v6.1.7+ / v7.0.1+)

Several callback hooks are optional: the dispatcher only invokes them when the handler defines the method, and emitters call CallbackManager.has_hook("...") to skip event construction entirely when nothing listens. This preserves the zero-cost invariant for legacy handlers.

on_llm_pre_call (v6.1.7) — wire-accurate prompt logging

LLMStartEvent fires BEFORE system-message consolidation — useful for debugging hydration. LLMPreCallEvent fires AFTER consolidation and tool binding, BEFORE the provider call — useful for gateway trace events, compliance capture, and replay fixtures.

from symfonic.core.contracts.callbacks import LLMPreCallEvent

class PromptAuditor:
    async def on_llm_pre_call(self, event: LLMPreCallEvent) -> None:
        # event.messages: post-consolidation list the provider sees
        # event.tools: resolved tool descriptors (name, description, schema)
        # event.invocation_params: provider-native kwargs (temperature, etc.)
        print(f"[wire] model={event.model} tools={len(event.tools)}")

Fields: model, messages: list[BaseMessage], tools: list[dict], run_id: str, invocation_params: dict. LLMPreCallEvent.messages contains the single leading SystemMessage (when any system content is present) followed by the non-system messages in order.

on_tool_routing_decision (v7.0.1) — wire-level tool narrowing

Fires when FrameworkConfig.tool_routing_mode != "off". Tells you which tools would (observe) or will (enforce) bind on this turn.

from symfonic.agent.triage.events import ToolRoutingDecisionEvent

class RoutingAuditor:
    async def on_tool_routing_decision(self, event: ToolRoutingDecisionEvent) -> None:
        print(
            f"[routing] {event.mode} intent={event.intent_label} "
            f"total={event.total_tools} narrowed={len(event.narrowed_tool_names)} "
            f"saved_tokens~{event.saved_tokens_estimate}"
        )

Use observe mode to collect telemetry before flipping to enforce — this is the recommended adoption path.

on_intent_classified (v7.0) — intent filter verdict

Fires as an ExtensionEvent(type="intent_classified") when intent_filter_mode != "off". Subscribe via the generic ExtensionEvent dispatch pattern if your handler supports one.

on_fabrication_detected (v7.0) — per-finding telemetry

Fires as an ExtensionEvent(type="fabrication_detected") when the fabrication detector finds a hit in observe, revise, or refuse mode. In refuse mode an ExtensionEvent(type="fabrication_blocked") also fires when the draft is replaced with a safe envelope.

FrameworkObservabilityHook: System-Level Metrics

For framework-level events (hydration latency, consolidation ops, routing, token budget), use FrameworkObservabilityHook:

from symfonic.agent.hooks import LoggingFrameworkHook
from symfonic.agent.config import FrameworkConfig

# LoggingFrameworkHook logs all framework events via Python logging
agent = SymfonicAgent(
    model_provider=provider,
    config=FrameworkConfig(observability_hook=LoggingFrameworkHook()),
)

The FrameworkObservabilityHook protocol defines these events:

Hook Method	Fires When
on_hydration_complete	Memory context injected into system prompt
on_consolidation_complete	Memory extraction committed to graph
on_routing_complete	Lazy tool routing filters tools for query
on_token_budget_computed	Token savings calculated from tool filtering

Streaming with FastAPI SSE

Combine stream_typed() with FastAPI's StreamingResponse for Server-Sent Events:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from symfonic.core.contracts.types import TextDeltaEvent
import json

app = FastAPI()

async def sse_generator(query: str, tenant_id: str):
    scope = FrameworkTenantScope(tenant_id=tenant_id)
    async for event in agent.stream_typed(query, scope=scope):
        event_type = type(event).__name__
        data = json.dumps(event.model_dump())
        yield f"event: {event_type}\ndata: {data}\n\n"

@app.post("/stream")
async def stream_endpoint(query: str, tenant_id: str):
    return StreamingResponse(
        sse_generator(query, tenant_id),
        media_type="text/event-stream",
    )

Or use the built-in router which handles SSE automatically:

from symfonic.agent.fastapi.router import create_agent_router

router = create_agent_router(agent, prefix="/api/v1")
app.include_router(router)
# SSE endpoint available at POST /api/v1/stream/typed

Next Steps

Your agent streams responses and you can observe every step. Level 4 shows how to inject domain-specific behavior with plugins.

Next: Plugins & Domains

See also: the Feature Catalog Observability cluster for the full event taxonomy and which release introduced each hook.