Skip to content

symfonic.core.streaming.adapters.openai_adapter

openai_adapter

OpenAI stream adapter -- converts raw OpenAI SSE to StreamEvent.

openai_adapter async 

openai_adapter(
    raw_stream: AsyncIterator[Any],
) -> AsyncIterator[StreamEvent]

Convert raw OpenAI stream chunks to StreamEvent.

Source code in src/symfonic/core/streaming/adapters/openai_adapter.py 
27
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
async def openai_adapter(
    raw_stream: AsyncIterator[Any],
) -> AsyncIterator[StreamEvent]:
    """Convert raw OpenAI stream chunks to StreamEvent."""
    async for chunk in raw_stream:
        choices = _get(chunk, "choices", [])

        if not choices:
            for evt in _handle_no_choices(chunk):
                yield evt
            continue

        choice = choices[0] if isinstance(choices, list) else choices
        delta = _get(choice, "delta", {})
        finish = _get(choice, "finish_reason")

        if finish:
            yield MessageEndEvent(message_id="", stop_reason=finish)
            continue

        yielded = False

        content = _get(delta, "content")
        if content:
            yield TextDeltaEvent(text=content)
            yielded = True

        tool_calls = _get(delta, "tool_calls")
        if tool_calls:
            for tc in tool_calls:
                for evt in _handle_tool_call(tc):
                    yield evt
            yielded = True

        if not yielded and delta:
            yield make_extension_event("openai.delta", delta)