Skip to content

symfonic.platform.transport_stream

transport_stream

Final-only publication for the two native SSE projections.

plain_events async 

plain_events(stream: AsyncIterator[Any], prepared: Any) -> AsyncIterator[str]

Emit text deltas; a valid final may publish before terminal success.

Source code in src/symfonic/platform/transport_stream.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
async def plain_events(
    stream: AsyncIterator[Any], prepared: Any,
) -> AsyncIterator[str]:
    """Emit text deltas; a valid final may publish before terminal success."""
    deltas = ""
    emitted_visible = False
    citations = CitationTagFilter()
    try:
        async for event in stream:
            kind = kind_of(event)
            if kind == "text_delta":
                text = getattr(event, "text", None)
                if isinstance(text, str) and text:
                    deltas += text
                    visible = citations.feed(text)
                    if visible:
                        emitted_visible = True
                        yield f"data: {json.dumps({'chunk': visible})}\n\n"
            elif kind == "done":
                try:
                    answer = visible_answer(final_answer_of(event, deltas))
                except UnusableFinal:
                    yield f"data: {json.dumps({'error': _UNUSABLE})}\n\n"
                    yield "data: [DONE]\n\n"
                    return
                if prepared is not None:
                    await prepared.publish(answer)
                tail = citations.flush()
                if tail:
                    emitted_visible = True
                    yield f"data: {json.dumps({'chunk': tail})}\n\n"
                if not emitted_visible:
                    yield f"data: {json.dumps({'chunk': answer})}\n\n"
                yield "data: [DONE]\n\n"
                return
            elif kind in _TERMINAL:
                yield f"data: {json.dumps({'error': _UNUSABLE})}\n\n"
                yield "data: [DONE]\n\n"
                return
        yield f"data: {json.dumps({'error': _UNUSABLE})}\n\n"
        yield "data: [DONE]\n\n"
    finally:
        await _close(stream)

typed_events async 

typed_events(stream: AsyncIterator[Any], prepared: Any) -> AsyncIterator[str]

Forward typed events while delaying only ambiguous citation boundaries.

Source code in src/symfonic/platform/transport_stream.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
async def typed_events(
    stream: AsyncIterator[Any], prepared: Any,
) -> AsyncIterator[str]:
    """Forward typed events while delaying only ambiguous citation boundaries."""
    deltas = ""
    citations = BufferedCitationStream()
    try:
        try:
            async for event in stream:
                kind = kind_of(event)
                text = getattr(event, "text", None)
                if kind == "done":
                    try:
                        answer = visible_answer(final_answer_of(event, deltas))
                    except UnusableFinal:
                        yield f"data: {json.dumps({'type': 'error', 'error': _UNUSABLE})}\n\n"
                        yield "data: [DONE]\n\n"
                        return
                body: dict[str, Any] = {"type": kind}
                for field in ("text", "index"):
                    value = getattr(event, field, None)
                    if value is not None:
                        body[field] = value
                if kind == "error":
                    body.pop("text", None)
                if kind == "error" or getattr(event, "error", None) is not None:
                    body["error"] = _UNUSABLE
                if kind == "done" or kind in _TERMINAL:
                    if kind == "done":
                        body["text"] = answer
                    ready = citations.finish()
                    if kind == "done" and prepared is not None:
                        await prepared.publish(answer)
                    for item in ready:
                        yield _ready_wire(item)
                    yield f"data: {json.dumps(body)}\n\n"
                    yield "data: [DONE]\n\n"
                    return
                if kind in {"text_delta", "thinking"} and isinstance(text, str):
                    if kind == "text_delta":
                        deltas += text
                    for item in citations.text(kind, text, body):
                        yield _ready_wire(item)
                else:
                    for item in citations.other(body):
                        yield _ready_wire(item)
            for item in citations.finish():
                yield _ready_wire(item)
            yield f"data: {json.dumps({'type': 'error', 'error': _UNUSABLE})}\n\n"
            yield "data: [DONE]\n\n"
        except CitationBufferOverflow:
            yield f"data: {json.dumps({'type': 'error', 'error': _UNUSABLE})}\n\n"
            yield "data: [DONE]\n\n"
    finally:
        await _close(stream)