EasyStreamingPython 3

Streaming Token Assembler

Assemble interleaved streaming token deltas into complete message text.

25m3 sample tests5 hidden tests

Assemble partial streaming deltas into completed message text.

Requirements

  • Define TokenAssembler.
  • apply(event) processes one event dict.
  • Each event has id, optional delta, and optional final.
  • Append delta text to that message ID.
  • If final is true, mark the message complete and return its full text.
  • If not final, return None.
  • get(message_id) returns current assembled text or None.
  • Ignore deltas for IDs already finalized.

Example

python
1a = TokenAssembler() 2assert a.apply({"id": "m", "delta": "hel"}) is None 3assert a.apply({"id": "m", "delta": "lo", "final": True}) == "hello"

Constraints

  • Support interleaved message IDs.
  • Don't drop empty-string deltas.
  • Finalizing twice should be idempotent.

Editor