Skip to content

symfonic.core.streaming

streaming

symfonic.core.streaming -- Stream event processing and adapters.

ActivationEvent

Bases: BaseModel

A single memory node activated during spreading activation.

ConsolidationScheduledEvent

Bases: BaseModel

Emitted after consolidation is scheduled (background task, not yet complete).

EventTranspiler 

EventTranspiler()

Converts raw LangGraph astream_events(v2) into typed StreamEvents.

Usage::

transpiler = EventTranspiler()
async for event in transpiler.transpile(runtime.astream_events("v2")):
    # event is a TimestampedEvent with ms precision
Source code in src/symfonic/core/streaming/transpiler.py 
58
59
60
def __init__(self) -> None:
    self._start_time: float = 0.0
    self._active_tool_calls: dict[str, str] = {}  # call_id -> tool_name

transpile async 

transpile(
    raw_events: AsyncIterator[dict[str, Any]],
) -> AsyncIterator[TimestampedEvent]

Transform raw LangGraph events into typed TimestampedEvents.

Source code in src/symfonic/core/streaming/transpiler.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
async def transpile(
    self,
    raw_events: AsyncIterator[dict[str, Any]],
) -> AsyncIterator[TimestampedEvent]:
    """Transform raw LangGraph events into typed TimestampedEvents."""
    self._start_time = time.monotonic()

    async for raw in raw_events:
        event_type = raw.get("event", "")
        data = raw.get("data", {})

        for stream_event in self._map_events(event_type, data, raw):
            elapsed = (time.monotonic() - self._start_time) * 1000
            yield TimestampedEvent(event=stream_event, elapsed_ms=elapsed)

ExtensionEvent

Bases: BaseModel

Vendor-specific or custom event. Never dropped, never raises.

ExtractionFilter 

ExtractionFilter()

Stateful streaming filter that strips extraction blocks from text.

Usage::

filt = ExtractionFilter()
for chunk in text_chunks:
    clean = filt.feed(chunk)
    if clean:
        yield TextDeltaEvent(text=clean)
# After the loop ends:
remainder = filt.flush()
if remainder:
    yield TextDeltaEvent(text=remainder)
ops = filt.get_extracted_ops()
Source code in src/symfonic/core/streaming/extraction_filter.py 
76
77
78
79
80
def __init__(self) -> None:
    self._buffer: str = ""
    self._ops: list[dict[str, Any]] = []
    # Whether we are currently inside a (potentially partial) tag.
    self._in_tag: bool = False

feed 

feed(text: str) -> str

Accept the next text chunk and return the clean portion.

The returned string is guaranteed to contain no extraction block content. It may be shorter than text, or empty if the chunk is (or partially is) an extraction block.

Parameters:

Name Type Description Default
text str

Raw text chunk from the LLM stream.

 required

Returns:

Type Description
str

Clean text safe to emit as a TextDeltaEvent. May be empty.
Source code in src/symfonic/core/streaming/extraction_filter.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def feed(self, text: str) -> str:
    """Accept the next text chunk and return the clean portion.

    The returned string is guaranteed to contain no extraction block
    content. It may be shorter than ``text``, or empty if the chunk
    is (or partially is) an extraction block.

    Args:
        text: Raw text chunk from the LLM stream.

    Returns:
        Clean text safe to emit as a TextDeltaEvent. May be empty.
    """
    self._buffer += text
    return self._process_buffer()

flush 

flush() -> str

Return any remaining buffered text that is not part of a tag.

Call this after the stream ends. If the buffer contains a partial tag that never completed, the partial tag text is returned as-is (it is not a real extraction block).

Returns:

Type Description
str

Any remaining clean text. May be empty.
Source code in src/symfonic/core/streaming/extraction_filter.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def flush(self) -> str:
    """Return any remaining buffered text that is not part of a tag.

    Call this after the stream ends. If the buffer contains a partial
    tag that never completed, the partial tag text is returned as-is
    (it is not a real extraction block).

    Returns:
        Any remaining clean text. May be empty.
    """
    remainder = self._buffer
    self._buffer = ""
    self._in_tag = False
    # One final pass: if a complete block is in the remainder, strip it.
    if remainder:
        remainder = self._strip_complete_blocks(remainder)
    return remainder

get_extracted_ops 

get_extracted_ops() -> list[dict[str, Any]]

Return all ops parsed from extraction blocks encountered so far.

Returns:

Type Description
list[dict[str, Any]]

List of op dicts (may be empty).
Source code in src/symfonic/core/streaming/extraction_filter.py 
120
121
122
123
124
125
126
def get_extracted_ops(self) -> list[dict[str, Any]]:
    """Return all ops parsed from extraction blocks encountered so far.

    Returns:
        List of op dicts (may be empty).
    """
    return list(self._ops)

GraphEdgeCreatedEvent

Bases: BaseModel

Emitted when a new edge is written during consolidation.

GraphNodeCreatedEvent

Bases: BaseModel

Emitted when a new node is written to the graph during consolidation.

MemoryExtractedEvent

Bases: BaseModel

Emitted when an extraction block is parsed from the stream.

MessageEndEvent

Bases: BaseModel

End of message.

MessageStartEvent

Bases: BaseModel

Start of a new message.

ResponseCompleteEvent

Bases: BaseModel

Final clean response text — equivalent of AgentResponse.final_response.

TextDeltaEvent

Bases: BaseModel

Incremental text chunk from LLM.

ThinkingDeltaEvent

Bases: BaseModel

Incremental thinking/reasoning chunk.

TimestampedEvent dataclass 

TimestampedEvent(event: StreamEvent, elapsed_ms: float)

StreamEvent wrapped with elapsed milliseconds from stream start.

to_dict 

to_dict() -> dict[str, Any]

Serialize for SSE transmission.

Source code in src/symfonic/core/streaming/transpiler.py 
39
40
41
42
43
44
45
def to_dict(self) -> dict[str, Any]:
    """Serialize for SSE transmission."""
    return {
        "elapsed_ms": round(self.elapsed_ms, 1),
        "event_type": type(self.event).__name__,
        "data": _event_to_dict(self.event),
    }

ToolCallDeltaEvent

Bases: BaseModel

Incremental tool call arguments.

ToolCallStartEvent

Bases: BaseModel

Tool call initiated by LLM.

ToolResultEvent

Bases: BaseModel

Result returned from tool execution.

UsageEvent

Bases: BaseModel

Token usage report.