Streaming the real model

In lesson 10.2, Model::reply started a task, and the task called ask, which waited for the whole answer and then sent it down the channel as a single piece. Ask the real model for a long story and the screen sits still for as long as the whole answer takes to write, then fills all at once. Meanwhile the model was writing the answer a word at a time, and you saw none of it.

The channel is already built for pieces: answer() in main.rs prints each one the moment it arrives. So nothing in main.rs changes in this lesson. All the work is in replier.rs, in how the task reads the reply.

The request gains one field, stream: bool, and Chatterbox sets it to true. That tells the service: do not wait until you are finished, send each piece as you write it. The reply then stops being one JSON object. It becomes server-sent events: plain lines of text that keep arriving over the same connection, one event at a time. A short answer looks like this (spaces matter; each piece of text starts with its own space):

data: {"choices":[{"delta":{"role":"assistant"}}]}

data: {"choices":[{"delta":{"content":"Hello"}}]}

data: {"choices":[{"delta":{"content":" there"}}]}

: keep-alive

data: [DONE]

Four facts cover all Chatterbox needs:

• An event is a line starting with data:, followed by some JSON. A blank line separates one event from the next.
• Instead of message, each piece has a delta (the word means "the change"): just the new text since the last piece.
• Some pieces carry no text at all. The first one often only says who is speaking. So content is an Option<String> in the new Delta struct.
• A line starting with : is a comment, which servers send to keep a quiet connection open. data: [DONE], which is not JSON at all, means the answer is complete. That one is the chat API's own convention, not a rule of server-sent events.

Every line of the stream means one of three things, so Chatterbox names them with an enum (lesson 5.3):

• Event::Piece(String): more of the answer, to print and to send down the channel.
• Event::Done: the answer is complete.
• Event::Other: anything else. A blank line, a comment, a piece with no text, or a line that is not valid JSON. Chatterbox skips it.

The function that decides is parse_event(line: &str) -> Event. It takes off the data: at the front with strip_prefix (which gives back an Option: Some of the rest, or None if the line does not start that way), trims the space after it, checks for [DONE], then lets serde read the JSON into the new structs Chunk, ChunkChoice and Delta. You write it yourself at the end of this lesson.

Here is the catch: the stream does not arrive line by line. response.chunk().await? gives you the next chunk of the reply: whatever bytes have arrived so far, as Some(bytes), or None once the reply is over. A chunk can be any size. It can hold three events, or half of one. It can end in the middle of a line.

Worse, it can end in the middle of a letter. Remember from lesson 2.5 that text is stored as UTF-8 bytes, and that a letter outside plain English takes more than one: é is 2 bytes, ☕ is 3. The network knows nothing about letters: it happily delivers the first byte of é in one chunk and the second byte in the next. What happens if you try to read the first chunk as text on its own?

The gentler cousin, String::from_utf8_lossy, never fails: it swaps any bytes that are not a whole letter for the replacement character �. That sounds safer and is worse, because the damage is silent. Read each chunk on its own and the reader sees caf� then another �, where the model wrote café:

$ cargo run
4 letters, 5 bytes
first chunk alone:  caf�
second chunk alone: �
both, together:     café

The fix is a waiting room. Chatterbox keeps everything that has arrived in a Vec<u8> called pending: bytes, not text, so half a letter can sit there safely. After each chunk it asks: is there a whole line in here yet? A line ends with a newline byte, written b'\n' (the b makes it a single byte, a u8, rather than a char).

• pending.iter().position(|&b| b == b'\n') finds the first newline and gives back its index as an Option. The & in |&b| is a pattern (lesson 5.5) that copies the byte out of the reference the iterator hands over.
• pending.drain(..=end) takes the bytes from the start up to and including end out of the Vec (..= is a range that includes its last number), and collect() gathers them into a new Vec<u8>. Whatever came after the newline stays behind for next time.
• Only now, with a whole line in hand, does from_utf8_lossy turn it into text. A whole line always holds whole letters. In UTF-8, every byte of a multi-byte letter is 0x80 or above, and the newline byte is 0x0A, so a newline never appears inside a letter and splitting at one can never cut a letter in half.

The inner while let (lesson 5.5) matters: one chunk may hold several lines, and they should all be read before waiting for more.

A tempting shortcut is to collect the drained bytes straight into a String. The compiler refuses:

error[E0277]: a value of type `String` cannot be built from an iterator over elements of type `u8`
 --> src/main.rs:5:50
  |
5 |         let line: String = pending.drain(..=end).collect();
  |                                                  ^^^^^^^ value of type `String` cannot be built from `std::iter::Iterator<Item=u8>`
  |
  = help: the trait `FromIterator<u8>` is not implemented for `String`
  = help: `String` implements trait `FromIterator<A>`:
            FromIterator<&char>
            FromIterator<&std::ascii::Char>
            FromIterator<&str>
            FromIterator<Box<str, A>>
            FromIterator<Cow<'_, str>>
            FromIterator<String>
            FromIterator<char>
            FromIterator<std::ascii::Char>
note: the method call chain might not have had the expected associated types
 --> src/main.rs:5:36
  |
2 |     let mut pending: Vec<u8> = Vec::new();
  |                                ---------- this expression has type `Vec<u8>`
...
5 |         let line: String = pending.drain(..=end).collect();
  |                                    ^^^^^^^^^^^^^ `Iterator::Item` is `u8` here
note: required by a bound in `collect`
 --> /rustc/b940084d7eb6a299eb4bfeb8e34901bc051e7ac4/library/core/src/iter/traits/iterator.rs:2073:4

In replier.rs, ask becomes stream. It still sends the request with .send().await? and checks the status with .error_for_status()?, but then, instead of reading the whole body as JSON, it loops: while let Some(bytes) = response.chunk().await?. Each chunk goes onto the end of pending; each whole line goes through parse_event. A Piece goes straight down the channel with tx.send(Ok(text)).await, and if that send fails, nobody is listening any more, so the function stops. Done ends the answer.

The channel now has room for 16 waiting pieces instead of 1. Reading the network pauses only if 16 pieces sit unread, the pause from lesson 10.2 that stops a fast writer drowning a slow reader.

One small rule closes it: finished(sent_any). A stream that ends, with [DONE] or by simply closing, without a single piece of text is not an answer. It is ChatError::Empty, the same error a reply with no choices was in lesson 9.2, and main.rs already prints it and carries on.

The run below is the proof. The course build starts a stand-in server that streams the way a real service does, one word per event, and it is deliberately unkind: it cuts every event into two chunks, right after the first byte of a multi-byte letter when there is one. The message has three such letters. Every one survived.

Sam
café ☕ naïve
/quit

Chatterbox v1.1 (replies by: real)
What should I call you?
Hi Sam! I'm Chatterbox. Ask me anything.
bot: mock model saw 1 message; last: café ☕ naïve
bot: bye, Sam!
messages so far: 1
saved 2 messages to chatterbox.json