Skip to content
NextiaNextia Learning

Stream responses

  • Lesson 15
  • 45 min
  • Module 5: Conversations and streaming
Lesson 15 of 19 · Building Reliable Applications with LLM APIsStream responses
Course overview

1. Start with one model request

  1. What an LLM API is
  2. A first run without an account
  3. Request anatomy
  4. The application boundary

2. Prompting as an engineering task

  1. Specify the task
  2. Provide context safely
  3. Compare prompts

3. Structured outputs and validation

  1. Define a schema
  2. Handle incomplete results
  3. Validate meaning

4. Tool calling safely

  1. Describe a tool
  2. Execute in application code
  3. Bound the loop

5. Conversations and streaming

  1. Manage conversation state
  2. Stream responses
  3. Communicate failure

6. Reliability and cost controls

  1. Handle provider limits
  2. Measure usage and cost
  3. Test and package

7. Case studies

  1. Extract product recalls into structured data
  2. Answer questions with tools over open data
  3. Classify banking requests with an LLM

What you will be able to do

Show the assistant’s draft while it arrives, so that the agent sees progress and can stop it. Make sure that only a complete answer, checked by both gates, can ever become data.

Before you startWhat to know and to have readyAbout 45 minutes

What you need to know

What you need on your computer

A phone number, digit by digit

Section titled “A phone number, digit by digit”

A colleague reads you a phone number over a bad line: “Zero, two, zero, seven…”. You write each digit as it comes. That is useful: you are ready sooner, and you can see that the call is going well. But you do not dial after four digits. “0207” is not a short phone number. It is part of a longer one, and you only know which one at the end.

A model’s answer can arrive in the same way, piece by piece. This lesson is about the two halves of that idea: show the pieces early, and never act on them until the end.

The problem: a long wait with nothing on screen

Section titled “The problem: a long wait with nothing on screen”

Until now, analyse waited for the whole answer, then printed it. For T-64704 (a kettle with a burning smell), the structured answer of Termchat-smallThe name that Larkfield’s code gives to a small, low-cost hosted model: gpt-6-luna on Azure. The code uses the name, so the model can change without a code change.Learn more What an LLM API isBuilding Reliable Applications with LLM APIs took 1.45 s. For other tickets it took up to 4 or 5 seconds (the Termp95The 95th percentile: the time that 95 of 100 calls stay under. It shows how long the slow calls take, which the median hides.Learn more Measure practical performanceLLMs: How They Work and How to Choose One latency of chat-small with prompt v2 was 4.34 s). During that time, the help-desk screen shows nothing. TermOmarLarkfield’s product manager. He asks the business questions: what a change costs, and what it gives the shop. worries that an agent will think the screen is frozen and click again. A second click is a second request, and a second bill.

Streaming is the answer to this problem. In a streamed request, the provider sends the answer in small pieces (also called chunks) as the model writes them, over one open connection. The application can show each piece at once. You ask for it with one field in the request: "stream": true.

TermAmiraLarkfield’s developer. She writes the code of the help-desk tools in the courses. recorded three real streams, with the time of every piece. Here is when the text of each one arrived:

  • chat-small, T-64704
  • chat-strong, T-64704
  • chat-small, T-80005
Three real streams with prompt v2 and structured output, recorded on Azure (chat-small is gpt-6-luna, chat-strong is gpt-6.1-sol). Each line is flat while nothing has arrived, then rises as the pieces come.
Text description of the chart

A line chart with three lines. The horizontal axis is the time since the request was sent, from 0 to 2.6 seconds. The vertical axis is the number of characters received. Every line stays at zero for a long time, then rises steeply. chat-small, T-64704: 106 pieces and 453 characters; the first piece at 1.275 s, the last at 1.554 s, the stream closed at 1.597 s. chat-strong, T-64704: 87 pieces and 408 characters; the first piece at 1.883 s, the last at 2.379 s, the stream closed at 2.42 s. chat-small, T-80005: 90 pieces and 382 characters; the first piece at 2.329 s, the last at 2.44 s, the stream closed at 2.506 s.

Look at the flat part. For chat-small on T-64704, the first piece came at 1.275 s, and the last at 1.554 s. So about 80% of the wait was before the first piece. For T-80005 it was 93%. These models reason before they write, and the reasoning is not streamed. Streaming does not make the answer faster. It shows the answer as soon as the model starts writing, and it shows that the request is alive.

The pieces are not words, and not data

Section titled “The pieces are not words, and not data”

Here are the first nine pieces of that stream, exactly as they arrived:

Piece Time (s) Text
1 1.275 {"
2 1.275 language
3 1.275 ":"
4 1.275 en
5 1.275 ","
6 1.275 team
7 1.275 ":"
8 1.275 w
9 1.276 arranty

The pieces follow the model’s tokens, not words or fields. After piece 8, the team is w. After piece 9, it is warranty. A piece can end in the middle of a word, a number or an order ID. T-80005’s stream sent LK, -, 266 and 415 as four pieces.

So text that is still arriving is partial text: a piece of an answer whose end you do not know yet. It is not data. You cannot parse it, and if you could, you could not trust it. Only when the stream ends normally do you have the complete answer. Then, and only then, it goes through the two gates of Module 3.

The idea: show early, check at the end

Section titled “The idea: show early, check at the end”

The picture follows one streamed answer. Every path ends in one of four states: valid, rejected, cancelled or interrupted.

A flow from top to bottom. Request sent with stream: true; waiting, no piece yet. A piece arrives. Show it as a draft, marked not checked. How did the stream end? More pieces: back to 'a piece arrives'. Normal end: complete text, then the two gates, schema then rules: pass gives valid, may be stored; fail gives rejected, a person checks. Agent pressed Stop: cancelled. Connection lost, or finish_reason length: interrupted. Cancelled and interrupted both lead to: partial text is not data, not checked, never stored.
The streaming lifecycle. The draft is shown while it arrives, but only a complete answer reaches the gates. Dashed boxes are endings without data.
Text description of the diagram
  1. The request is sent with stream: true. The screen waits: no piece yet.
  2. A piece arrives. The screen shows it as part of a draft, marked “not checked”.
  3. How did the stream end? If more pieces come, go back to step 2.
  4. Normal end: the text is complete. It goes through the two gates, schema then rules. If both pass, the answer is valid and may be stored. If a gate fails, it is rejected and a person checks the ticket.
  5. The agent pressed Stop: the stream is cancelled.
  6. The connection was lost, or the stream ended with finish_reason “length”: the stream is interrupted.
  7. Cancelled and interrupted answers are partial text. They are not checked and never stored.

Two of the endings need a word more. Cancel means that the agent pressed Stop. The application stops reading and closes the connection. A provider can still bill the tokens that the model wrote before the stop; check your provider’s documentation. An interrupted stream ends without a normal end: the network breaks, or the model hits the token limit (finish_reason: "length"). The project treats both in the same way: the text that arrived stays on the screen as a broken draft, and it never goes further.

A tiny example: the same stream, three endings

Section titled “A tiny example: the same stream, three endings”

The project’s stream_answer() reads the pieces, shows each one, and checks a cancel() function before each piece. It returns a StreamResult with the text, the pieces and how the stream ended. Here it is on the recorded stream, run to the end:

Terminal window
python -m assistant.stream T-64704

The output (the real stream of chat-small, replayed with its recorded timing):

{"language":"en","team":"warranty","needs_human":true,"reason":"A problem developed during use, so this is a warranty issue; the heat, burning smell and scorch mark are safety concerns (H1).","confidence":"high","order":null,"reply":"I’m sorry this happened, and you did the right thing by stopping use of the kettle. Please keep it unplugged and do not use it again. A member of our team will contact you about the next steps and your safety concern."}


106 pieces, first after 1.27 s, all after 1.55 s
complete | valid

The text appears piece by piece in your terminal. Your times can differ by a few hundredths of a second. Now the agent presses Stop after 20 pieces:

Terminal window
python -m assistant.stream T-64704 --cancel-after 20
{"language":"en","team":"warranty","needs_human":true,"reason":"A problem


20 pieces, first after 1.27 s, all after 1.29 s
cancelled | not checked: partial text is not data

And the connection breaks after 40 pieces. This break is simulated: the option --cut-after stops the recorded stream on purpose.

Terminal window
python -m assistant.stream T-64704 --cut-after 40
{"language":"en","team":"warranty","needs_human":true,"reason":"A problem developed during use, so this is a warranty issue; the heat, burning smell and scorch mark are


40 pieces, first after 1.27 s, all after 1.40 s
interrupted: Connection lost (simulated). | not checked: partial text is not data

Every run ends with one status line. Only the first one says valid. The other two say what happened and that nothing was checked.

Add streaming to the project

Section titled “Add streaming to the project”

  1. Open assistant/providers.py. Add from typing import Iterator to the imports at the top.

  2. In MockProvider, give __init__ a speed setting, and add a stream() method below complete():

    assistant/providers.py (MockProvider)
        def __init__(self, folder: Path = RECORDINGS, speed: float = 0.0):
            self.speed = speed  # 1.0 replays a stream with its recorded timing; 0 replays it at once
            self.recordings = {}
            # ... the rest of __init__ stays the same
    

        def stream(self, request: dict) -> Iterator[tuple[float, str]]:
            """Yield (seconds since the request, text piece) as the recorded stream did."""
            entry = self._entry({**request, "stream": True})
            start = time.monotonic()
            for at, piece in entry["chunks"]:
                if self.speed:
                    time.sleep(max(0.0, at / self.speed - (time.monotonic() - start)))
                yield at, piece

    The mock looks up the request with "stream": true, because a streamed request is a different request.

  3. In OpenAICompatibleProvider, add a stream() method below complete():

    assistant/providers.py (OpenAICompatibleProvider)
        def stream(self, request: dict) -> Iterator[tuple[float, str]]:
            start = time.monotonic()
            with translate_errors():
                for chunk in self.client.chat.completions.create(**request, stream=True):
                    if chunk.choices and chunk.choices[0].delta.content:
                        yield round(time.monotonic() - start, 3), chunk.choices[0].delta.content
                    if chunk.choices and chunk.choices[0].finish_reason not in (None, "stop"):
                        # for example "length": the stream ended normally, but the answer was cut
                        raise ProviderError(f"The stream ended with finish_reason: {chunk.choices[0].finish_reason}.")

    The last check matters. A stream cut by the token limit closes like a normal stream. Without the check, the code would call a cut answer complete.

  4. Make the file assistant/stream.py:

    assistant/stream.py
    """Show the answer as it arrives, allow cancel, and check only the complete answer.
    

        python -m assistant.stream T-64704                     replays a recorded stream with its real timing
        python -m assistant.stream T-64704 --cancel-after 20   the agent presses Stop after 20 pieces
        python -m assistant.stream T-64704 --cut-after 40      the connection breaks after 40 pieces (simulated)
    """
    

    import argparse
    import sys
    import time
    from dataclasses import dataclass, field
    from typing import Callable
    

    from .providers import Completion, ProviderError
    

    

    @dataclass
    class StreamResult:
        text: str = ""
        complete: bool = False   # True only when the stream ended normally
        cancelled: bool = False
        error: str = ""
        pieces: list[tuple[float, str]] = field(default_factory=list)
    

        @property
        def first_piece_s(self) -> float | None:
            return self.pieces[0][0] if self.pieces else None
    

        @property
        def status(self) -> str:
            """How the stream ended: complete, cancelled, interrupted (cut after some pieces) or failed (no piece)."""
            if self.complete:
                return "complete"
            if self.cancelled:
                return "cancelled"
            return "interrupted" if self.pieces else "failed"
    

    

    def stream_answer(provider, request: dict, show: Callable[[str], None] = lambda piece: None,
                      cancel: Callable[[], bool] = lambda: False) -> StreamResult:
        result = StreamResult()
        try:
            for at, piece in provider.stream(request):
                if cancel():
                    result.cancelled = True
                    return result
                result.pieces.append((at, piece))
                result.text += piece
                show(piece)
        except ProviderError as error:
            result.error = str(error)  # what arrived before the error stays partial text
            return result
        result.complete = True
        return result
    

    

    def as_completion(result: StreamResult) -> Completion:
        """Only a complete stream becomes a Completion that validate() may check."""
        if not result.complete:
            raise ValueError("Partial text is not data: the stream did not finish.")
        data = {"choices": [{"message": {"content": result.text}, "finish_reason": "stop"}]}
        return Completion.from_response(data, result.pieces[-1][0] if result.pieces else 0.0)
    

    

    class CutAfter:
        """Simulated: the connection breaks after `pieces` pieces of a real (or recorded) stream."""
    

        def __init__(self, inner, pieces: int):
            self.inner, self.pieces = inner, pieces
    

        def stream(self, request: dict):
            for i, item in enumerate(self.inner.stream(request)):
                if i == self.pieces:
                    raise ProviderError("Connection lost (simulated).")
                yield item
    

    

    def main(argv: list[str] | None = None) -> int:
        from .config import Settings, make_provider
        from .context import build_request
        from .data import load_ticket
        from .orders import OrderBook
        from .schema import response_format
        from .validate import validate
    

        parser = argparse.ArgumentParser(prog="python -m assistant.stream")
        parser.add_argument("ticket_id", nargs="?", default="T-64704")
        parser.add_argument("--cancel-after", type=int, metavar="N", help="press Stop after N pieces")
        parser.add_argument("--cut-after", type=int, metavar="N", help="break the connection after N pieces (simulated)")
        args = parser.parse_args(argv)
    

        settings = Settings.from_env()
        provider = make_provider(settings)
        if hasattr(provider, "speed"):
            provider.speed = 1.0  # replay the recorded timing
        if args.cut_after is not None:
            provider = CutAfter(provider, args.cut_after)
        try:
            ticket = load_ticket(args.ticket_id)
        except KeyError as error:
            print(error.args[0], file=sys.stderr)
            return 2
        request = build_request(ticket, settings.model, "v2", response_format())
        shown = []
    

        def show(piece: str) -> None:
            print(piece, end="", flush=True)
            shown.append(piece)
    

        def stop() -> bool:  # the agent's Stop button, pressed after N pieces
            return args.cancel_after is not None and len(shown) >= args.cancel_after
    

        start = time.monotonic()
        result = stream_answer(provider, request, show=show, cancel=stop)
        first = f"first after {result.first_piece_s:.2f} s, " if result.pieces else ""
        print(f"\n\n{len(result.pieces)} pieces, {first}all after {time.monotonic() - start:.2f} s")
        if not result.complete:
            reason = f": {result.error}" if result.error else ""
            checked = "not checked: partial text is not data" if result.pieces else "no answer"
            print(f"{result.status}{reason} | {checked}")
            return 1
        verdict = validate(as_completion(result), ticket, OrderBook())
        checked = "valid" if verdict.valid else "rejected: " + ", ".join(p.code for p in verdict.problems)
        print(f"complete | {checked}")
        return 0 if verdict.valid else 1
    

    

    if __name__ == "__main__":
        sys.exit(main())

  5. Copy recordings/streams.jsonl from the snapshot end-of-m05 into your recordings folder, as in the previous lesson.

  6. Run the three commands of the tiny example and compare your output.

  7. Copy tests/test_stream.py from the snapshot end-of-m05 into your tests folder. Run python -m pytest.

Step 7 prints 65 passed, 1 deselected. The three new tests replay the recorded stream to the end, cancel it after 20 pieces, and cut it after 10 pieces (simulated). The last two check that as_completion() refuses the partial text.

The design puts the rule in one place. as_completion() is the only way from a stream to a Completion, and validate() takes only a Completion. So no code path can check, or store, the text of a stream that did not finish.

To watch the other recorded streams, set the model before the command. On macOS or Linux: ASSISTANT_MODEL=chat-strong python -m assistant.stream T-64704. On Windows, in PowerShell: $env:ASSISTANT_MODEL="chat-strong", then the command. Termchat-strongThe name that Larkfield’s code gives to a larger hosted model: gpt-6.1-sol on Azure. It is often more accurate than chat-small, and it costs about 14 to 18 times more per call.Learn more Request anatomyBuilding Reliable Applications with LLM APIs’s stream starts at 1.88 s and ends complete | valid. Remove the setting afterwards (Remove-Item Env:ASSISTANT_MODEL in PowerShell).

When to stream, and what it costs

Section titled “When to stream, and what it costs”

Stream when a person waits and watches: the help-desk screen, a chat. Do not stream for work that nobody watches, such as the nightly eval over 69 tickets. There, streaming only adds code.

Streaming has costs of its own:

  • The tokens cost the same. The model writes the same answer. Cancelling saves only the part that it had not written yet, if your provider stops billing at the cancel.
  • Usage needs a request. A plain stream does not report token usage. Many TermChat CompletionsThe request and response format of OpenAI’s chat API: a model name, a list of messages with roles, and settings. Azure, Ollama and many other providers accept the same format (“OpenAI-compatible”), so one SDK works with all of them.Learn more Request anatomyBuilding Reliable Applications with LLM APIs APIs send it at the end only if the request asks for it. The project’s stream does not ask, so Measure usage and cost logs the usage of normal calls.
  • More states to handle. A normal call succeeds or fails. A stream can also stop in the middle. The next lesson designs what the agent sees in each case.
  • TermStructured outputA request that sends the schema of the answer to the provider, so the model can write only an answer of that shape. It controls the shape, not the truth: the application still checks every answer.Learn more Define a schemaBuilding Reliable Applications with LLM APIs shows badly. The agent does not want to read {"language":"en","team":"w. A real screen would show only the reply field as it grows. In the T-64704 stream, the reply began at 1.395 s, only 0.20 s before the end. For short structured answers on a TermReasoning modelA model that writes hidden reasoning tokens before its answer. chat-small is one. Such a model often accepts only its default temperature, and its hidden tokens count against the output limit.Learn more Reasoning and multimodalityLLMs: How They Work and How to Choose One, a clear “Writing a draft…” state may help the agent as much as the text itself.

Your turn: when could you check T-80005?

Section titled “Your turn: when could you check T-80005?”

The recorded stream of T-80005 has 90 pieces. Run it with --cut-after 45. Before you run it, predict two things: which field is the stream in when it breaks, and what the status line says. Then answer: is there any piece number, before the end, after which the analysis could be checked safely?

Show the answer
{"language":"fr","team":"delivery","needs_human":false,"reason":"Le suivi est bloqué et la commande n’est pas arrivée : c’est une demande de livraison.","confidence":"high","order":{"order_id


45 pieces, first after 2.33 s, all after 2.40 s
interrupted: Connection lost (simulated). | not checked: partial text is not data

The stream broke inside the order object, before the order ID. And no, no earlier piece is safe. The JSON is not complete before the last piece, so gate 1 (the schema) cannot run. Even a field that looks finished, such as "team":"delivery", is safe only if you know that nothing later changes the meaning. The reply and the order come later, and the rules check them together. The only safe point is the normal end of the stream.

Common mistake

'Just close the brackets and use what we have'

Symptom: a stream stops early. TermTomásLarkfield’s AI engineer. He often suggests a quick shortcut, and the lesson shows why it fails. has an idea: “The JSON is almost there. Add the missing "} and parse it.” Amira tries it on the recorded T-64704 stream, stopped after 76 pieces. This is simulated from the real recording: the stream really had 106 pieces.

close_brackets.py
from assistant.context import build_request
from assistant.data import load_ticket
from assistant.orders import OrderBook
from assistant.providers import Completion, MockProvider
from assistant.schema import response_format
from assistant.stream import stream_answer
from assistant.validate import validate


ticket = load_ticket("T-64704")
request = build_request(ticket, "chat-small", "v2", response_format())
count = iter(range(1000))
result = stream_answer(MockProvider(), request, cancel=lambda: next(count) >= 76)  # Stop after 76 pieces
print(result.status, "after", len(result.pieces), "pieces")


repaired = result.text + '"}'  # Tomás: "just close the quote and the bracket"
data = {"choices": [{"message": {"content": repaired}, "finish_reason": "stop"}]}
verdict = validate(Completion.from_response(data, 0.0), ticket, OrderBook())
print("valid:", verdict.valid, "| needs_human:", verdict.analysis.needs_human)
print("reply:", verdict.analysis.reply)
cancelled after 76 pieces
valid: True | needs_human: True
reply: I’m sorry this happened, and you did the right thing by stopping use of the kettle.

Both gates pass. The reply reads like a finished sentence. But the complete reply went on: “Please keep it unplugged and do not use it again. A member of our team will contact you about the next steps and your safety concern.” The repaired draft drops the safety advice and the promise of contact, on a ticket about a burning smell.

Cause: partial text can look complete. The schema checks the form, and the rules check what they were written to check. Neither can know what the model would have written next. A repair invents an ending: here, the ending “nothing more”.

Fix: never repair, parse or store partial text. Show it as a broken draft, marked “not checked”, and offer a new attempt or a manual reply. In the project, as_completion() refuses a stream that did not finish, so the only path to validate() is a complete answer.

What it means for a real AI service: “almost complete” is not a kind of complete. Any data that drives a decision must come from an answer that ended normally and passed every check.

Knowledge check

Section titled “Knowledge check”

Check your understanding

In the recorded chat-small stream of T-64704, the first piece arrived at 1.275 s and the stream closed at 1.597 s. What did streaming improve, and what did it not improve?

Show the answer

It improved what the agent sees. The first text appears at 1.28 s, so the agent knows that the request is alive and can stop it. It did not make the answer faster. About 80% of the wait came before the first piece, while the model reasoned, and the whole answer still took about 1.6 s. The usable, checked answer is ready only at the end.

Check your understanding

A stream ends with finish_reason 'length'. The connection closed normally. Why does the project treat it as interrupted, and what would go wrong otherwise?

Show the answer

“length” means that the model hit the token limit and the answer was cut. The text is partial, even though the connection closed in the normal way. Without the check in OpenAICompatibleProvider.stream(), stream_answer() would mark the result complete, and as_completion() would pass it on with finish_reason: "stop". With structured output the cut JSON would usually fail gate 1 (the schema). But a cut plain-text answer could look finished, as the FailureCase shows.

Check your understanding

Predict: you run python -m assistant.stream T-80008 with the mock. What happens, and why?

Show the answer

The mock has no recorded stream for T-80008, so it raises RecordingNotFound before the first piece. stream_answer() catches it as a ProviderError. The result has no pieces, so its status is failed. The command prints 0 pieces, then failed: No recording for this request (key …) and no answer, and it ends with exit code 1. Nothing is checked or stored.

Recap

  • Streaming sends the answer in pieces as the model writes it ("stream": true). It shows progress and lets the agent stop. It does not make the answer faster: in the recorded streams, 78% to 93% of the wait came before the first piece.
  • Pieces follow tokens, not fields: the team arrived as w, then arranty. Text that is still arriving is partial text, and partial text is not data.
  • Every stream ends in one state: complete, cancelled (the agent pressed Stop), interrupted (the connection broke, or finish_reason was “length”) or failed (no piece at all).
  • Only a complete stream goes through the two gates. In the project, as_completion() refuses anything else, so no partial text can be checked or stored.
  • A repaired partial answer can pass both gates and still be wrong. The T-64704 reply, cut after 76 pieces (simulated), lost its safety advice.

Next, decide what the agent sees in each of these states, in words that help and without internal details.

Next: Communicate failure