LLM APIs stream tokens as server-sent events. The bytes arrive in arbitrary chunks (a line can be split across chunks). Parse them per the SSE rules:
\n or \r\n.field: value — strip one space after the colon if present. Fields: event, data, id.: are comments. A line without a colon is a field with an empty value.data lines in one event are joined with \n.{"event": type or "message", "data": ..., "id": last id seen in the stream or None}. event and data reset after dispatch; id persists.Implement parse_sse. Input: chunks: list[str]. Returns: list[dict].
def parse_sse(chunks: list[str]) -> list[dict]:
events = []
for chunk in chunks:
for line in chunk.split("\n"):
if line.startswith("data: "):
events.append({"event": "message", "data": line[6:], "id": None})
return events
Run executes the public tests. Submit grades against public and hidden tests.
LLM APIs stream tokens as server-sent events. The bytes arrive in arbitrary chunks (a line can be split across chunks). Parse them per the SSE rules:
\n or \r\n.field: value — strip one space after the colon if present. Fields: event, data, id.: are comments. A line without a colon is a field with an empty value.data lines in one event are joined with \n.{"event": type or "message", "data": ..., "id": last id seen in the stream or None}. event and data reset after dispatch; id persists.Implement parse_sse. Input: chunks: list[str]. Returns: list[dict].
def parse_sse(chunks: list[str]) -> list[dict]:
events = []
for chunk in chunks:
for line in chunk.split("\n"):
if line.startswith("data: "):
events.append({"event": "message", "data": line[6:], "id": None})
return events
Run executes the public tests. Submit grades against public and hidden tests.