What This Project Extends

Unit 4, Lesson 5 built a command-line chatbot with three interchangeable memory backends — manual state, previous_response_id chaining, and the Conversations API — all producing complete, non-streaming replies. This project extends that same chatbot to stream its replies token by token to the terminal, applying this unit's Lessons 2 and 3 directly to real, previously-written code, and demonstrates the specific interaction between streaming and each of the three memory mechanisms, which behaves slightly differently in each case, as this lesson will show concretely.

Reviewing What Needs to Change

Each of the three backends from Unit 4's project currently calls client.responses.create() synchronously and processes a complete response object. Converting each to streaming means adding stream=True, consuming the resulting event sequence (per this unit's Lesson 2), and — critically — capturing whatever state each backend needs for its next turn (the growing history for manual state, the response ID for chaining, nothing extra for the Conversations API) from the appropriate lifecycle event (per Lesson 3), since that information is no longer immediately available the way it was on a plain, non-streaming response object.

# backends.py — the shared streaming consumption helper all three backends will use

def stream_and_collect(stream, on_delta=None) -> tuple[str, "Response"]:
    """Consume a stream, returning the complete text and the final response object
    once streaming finishes — the response object is where the ID and usage live."""
    collected = ""
    final_response = None
    for event in stream:
        if event.type == "response.output_text.delta":
            collected += event.delta
            if on_delta:
                on_delta(event.delta)
        elif event.type == "response.completed":
            final_response = event.response
        elif event.type == "response.error":
            raise RuntimeError(f"Stream reported an error: {event.error}")
    return collected, final_response

This helper is exactly the StreamHandler pattern from Lesson 3, expressed as a plain function rather than a class for simplicity within this smaller project — it captures the two things every backend will need: the complete text (for display and for updating stored history) and the final response object (for the response ID, needed by the chaining backend specifically).

Updating the Manual-State Backend

class ManualStateBackend(ChatBackend):
    def __init__(self, conn):
        self.conn = conn
        self.history = load_state(conn, "manual", default=[])

    def send(self, user_message: str) -> str:
        self.history.append({"role": "user", "content": user_message})

        stream = client.responses.create(
            model="gpt-5.6-luna",
            instructions=INSTRUCTIONS,
            input=self.history,
            stream=True,
        )
        collected, _ = stream_and_collect(
            stream, on_delta=lambda chunk: print(chunk, end="", flush=True)
        )

        self.history.append({"role": "assistant", "content": collected})
        save_state(self.conn, "manual", self.history)
        return collected

    def reset(self) -> None:
        self.history = []
        clear_state(self.conn, "manual")

The only structural change from Unit 4's original version is replacing the single blocking client.responses.create() call and its response.output_text access with a streaming call consumed through stream_and_collect(), printing each chunk as it arrives via the on_delta callback while still ending up with the complete text needed to update self.history exactly as before. The manual-state backend doesn't need anything from the final response object itself (no response ID to track, since this backend doesn't chain), so the underscore in collected, _ = stream_and_collect(...) discards it.

Updating the Chaining Backend

class ChainingBackend(ChatBackend):
    def __init__(self, conn):
        self.conn = conn
        self.last_response_id = load_state(conn, "chain", default=None)

    def send(self, user_message: str) -> str:
        kwargs = {"model": "gpt-5.6-luna", "instructions": INSTRUCTIONS,
                  "input": user_message, "stream": True}
        if self.last_response_id:
            kwargs["previous_response_id"] = self.last_response_id

        stream = client.responses.create(**kwargs)
        collected, final_response = stream_and_collect(
            stream, on_delta=lambda chunk: print(chunk, end="", flush=True)
        )

        if final_response is None:
            raise RuntimeError("Stream completed without a response.completed event — "
                               "cannot continue the chain without a response ID.")

        self.last_response_id = final_response.id
        save_state(self.conn, "chain", self.last_response_id)
        return collected

    def reset(self) -> None:
        self.last_response_id = None
        clear_state(self.conn, "chain")

This backend is where the streaming-and-memory interaction matters most directly: unlike the non-streaming version from Unit 4, where response.id was available the instant the call returned, the streaming version can only obtain the response ID once the stream has fully completed and delivered its response.completed event — which is exactly why stream_and_collect() was designed to return the final response object alongside the collected text. The explicit if final_response is None check guards against a genuinely important failure mode: if a stream is somehow interrupted before response.completed ever fires (a network drop mid-stream, for instance), this backend would otherwise silently fail to update self.last_response_id, breaking the chain for the next turn without any obvious error — raising an explicit exception here surfaces that problem immediately rather than letting it manifest later as a confusing "the bot forgot" symptom.

Updating the Conversations API Backend

class ConversationBackend(ChatBackend):
    def __init__(self, conn):
        self.conn = conn
        self.conversation_id = load_state(conn, "conversation", default=None)
        if self.conversation_id is None:
            conversation = client.conversations.create(metadata={"source": "cli_chatbot"})
            self.conversation_id = conversation.id
            save_state(self.conn, "conversation", self.conversation_id)

    def send(self, user_message: str) -> str:
        stream = client.responses.create(
            model="gpt-5.6-luna",
            instructions=INSTRUCTIONS,
            conversation=self.conversation_id,
            input=user_message,
            stream=True,
        )
        collected, _ = stream_and_collect(
            stream, on_delta=lambda chunk: print(chunk, end="", flush=True)
        )
        return collected

    def reset(self) -> None:
        if self.conversation_id:
            client.conversations.delete(conversation_id=self.conversation_id)
        conversation = client.conversations.create(metadata={"source": "cli_chatbot"})
        self.conversation_id = conversation.id
        save_state(self.conn, "conversation", self.conversation_id)

Notice this backend, like the manual-state one, discards the final response object — the Conversations API tracks accumulated history server-side against the stable conversation_id, which doesn't change turn to turn, so unlike the chaining backend there's nothing new to capture from the completed response in order to continue correctly on the next turn.

Adding a Streaming-Aware Cost Tracker

Unit 4, Lesson 5's cost-tracking extension read response.usage directly from a completed, non-streaming response. Under streaming, usage is only available via the response.completed event's response object, exactly like the chaining backend's response ID — the same underlying lesson applies.

# backends.py — updated cost tracking, now reading usage from the streamed completion event

def stream_collect_and_cost(stream, on_delta=None) -> dict:
    collected = ""
    final_response = None
    for event in stream:
        if event.type == "response.output_text.delta":
            collected += event.delta
            if on_delta:
                on_delta(event.delta)
        elif event.type == "response.completed":
            final_response = event.response

    cost = estimate_cost(final_response.usage) if final_response else None
    return {"text": collected, "response": final_response, "cost": cost}

Any of the three backends can call stream_collect_and_cost() instead of the simpler stream_and_collect() shown earlier to also report a per-turn cost estimate, exactly as Unit 4's cost-tracking extension did for the non-streaming version — the extension's core logic (estimate_cost()) doesn't need to change at all, only where its input (usage) comes from.

Adding a "Stop Generating" Command

Lesson 1 of this unit highlighted early cancellation as one of streaming's distinct benefits — this project is a natural place to actually implement it, letting the user interrupt an in-progress response by pressing Ctrl+C during generation, and having the application handle that interruption gracefully rather than crashing.

# chatbot.py — handling an interrupt during streaming, rather than only between turns

def send_with_interrupt_handling(backend, user_message: str) -> str:
    try:
        return backend.send(user_message)
    except KeyboardInterrupt:
        print("\n[Generation stopped by user]")
        return "[response interrupted]"

For this to actually stop the underlying stream promptly (rather than merely stopping your own code from waiting on it further, while the server continues generating a response nobody will use), each backend's streaming call should also close the stream explicitly on interruption, following Lesson 2's stream.close() pattern:

def send(self, user_message: str) -> str:
    self.history.append({"role": "user", "content": user_message})
    stream = client.responses.create(
        model="gpt-5.6-luna", instructions=INSTRUCTIONS, input=self.history, stream=True,
    )
    try:
        collected, _ = stream_and_collect(
            stream, on_delta=lambda chunk: print(chunk, end="", flush=True)
        )
    except KeyboardInterrupt:
        stream.close()
        collected = "[response interrupted]"
        print("\n[stopped]")

    self.history.append({"role": "assistant", "content": collected})
    save_state(self.conn, "manual", self.history)
    return collected

Storing "[response interrupted]" (rather than nothing, or a partial fragment) as the assistant's turn in history keeps the conversation record honest about what actually happened, which matters for the model's own understanding of the conversation on subsequent turns — leaving a silent gap or a suspiciously truncated sentence in the history could otherwise confuse a later turn that references "what you just said."

Running and Comparing All Three Streaming Backends

With all three backends updated, running the same commands from Unit 4's project now demonstrates streaming behavior directly:

python chatbot.py --backend manual
python chatbot.py --backend chain
python chatbot.py --backend conversation

Each should now display replies progressively rather than all at once — worth directly comparing side by side against the Unit 4 version (or a --no-stream flag added for comparison, toggling between the streaming and non-streaming code paths) to see Lesson 1's perceived-responsiveness argument made concrete in a project you built yourself, rather than only in the isolated examples earlier in this unit.

Adding a "Generating..." Indicator Before the First Token Arrives

Even with streaming, there's typically a brief delay between submitting a request and the first response.output_text.delta event arriving — the model still needs to begin processing before any output exists. Following Lesson 3's response.created event pattern, the chatbot can show a lightweight indicator during this initial gap, clearing it the moment real content starts arriving.

def stream_and_collect_with_indicator(stream) -> tuple[str, "Response"]:
    collected = ""
    final_response = None
    first_content_seen = False

    print("(generating...)", end="", flush=True)
    for event in stream:
        if event.type == "response.output_text.delta":
            if not first_content_seen:
                print("\r" + " " * 15 + "\r", end="")  # clear the indicator
                first_content_seen = True
            print(event.delta, end="", flush=True)
            collected += event.delta
        elif event.type == "response.completed":
            final_response = event.response
        elif event.type == "response.error":
            raise RuntimeError(f"Stream reported an error: {event.error}")

    return collected, final_response

This small addition directly applies Lesson 1's argument about the value of continuous evidence that something is happening: even a brief, simple "(generating...)" placeholder measurably improves the felt experience of the short gap before the first real token appears, compared to a terminal that shows nothing at all during that same interval.

Extending Toward a Simple Web Interface

The command-line project is deliberately minimal so the streaming mechanics stay front and center, but it's worth sketching how the same backend logic would extend to a simple web-based chat interface, connecting this project back to Lesson 2's Flask/SSE example.

from flask import Flask, Response, request

app = Flask(__name__)
conn = init_db()
active_backend = ChainingBackend(conn)  # or select based on a request parameter

@app.route("/chat", methods=["POST"])
def chat():
    user_message = request.json["message"]

    def generate():
        kwargs = {"model": "gpt-5.6-luna", "instructions": INSTRUCTIONS,
                  "input": user_message, "stream": True}
        if active_backend.last_response_id:
            kwargs["previous_response_id"] = active_backend.last_response_id

        stream = client.responses.create(**kwargs)
        for event in stream:
            if event.type == "response.output_text.delta":
                yield f"data: {event.delta}\n\n"
            elif event.type == "response.completed":
                active_backend.last_response_id = event.response.id
                save_state(conn, "chain", active_backend.last_response_id)
        yield "data: [DONE]\n\n"

    return Response(generate(), mimetype="text/event-stream")

The essential logic — capturing the response ID from response.completed to continue the chain correctly — is identical to the command-line version; only the delivery mechanism (an SSE-formatted HTTP response instead of print() to a terminal) differs, which is exactly the kind of separation of concerns worth aiming for: the memory-and-streaming logic from this unit and Unit 4 stays the same regardless of whether the final destination is a terminal or a browser.

Testing the Updated Backends

Following this course's consistent dependency-injection testing pattern, the streaming versions of each backend can be tested with a fake stream standing in for a real API call, verifying specifically that the response ID and history updates happen correctly once a stream completes.

def test_chaining_backend_updates_id_after_stream():
    class FakeEvent:
        def __init__(self, type_, **attrs):
            self.type = type_
            for k, v in attrs.items():
                setattr(self, k, v)

    class FakeResponse:
        def __init__(self, id_):
            self.id = id_

    def fake_stream():
        yield FakeEvent("response.output_text.delta", delta="Hi ")
        yield FakeEvent("response.output_text.delta", delta="there!")
        yield FakeEvent("response.completed", response=FakeResponse("resp_test_1"))

    import backends
    backends.client = type("FakeClient", (), {
        "responses": type("FakeResponses", (), {
            "create": staticmethod(lambda **kwargs: fake_stream())
        })()
    })()

    conn = init_db()
    backend = ChainingBackend(conn)
    reply = backend.send("Hello")

    assert reply == "Hi there!"
    assert backend.last_response_id == "resp_test_1"
    print("PASS: chaining backend correctly captures response ID from a streamed completion")

test_chaining_backend_updates_id_after_stream()

This test directly targets the exact bug class this lesson has repeatedly warned about — a chaining backend that fails to correctly capture the response ID from a streamed completion — making it something the project's own test suite catches automatically rather than something that could silently regress in a future edit to the streaming logic.

Troubleshooting Checklist for This Project

Replies print all at once instead of progressively. Check that flush=True is set on every print() call handling a delta chunk, and confirm the terminal or environment you're running in isn't itself buffering output in a way that defeats incremental display (some IDE-integrated terminals behave this way by default).

The chain backend "forgets" after an interruption. Confirm the interrupted-stream handling explicitly closes the stream and does not attempt to update last_response_id from a final_response that was never actually set — the if final_response is None guard shown earlier in this lesson exists specifically to catch this and fail loudly rather than silently corrupting the chain.

Cost estimates from stream_collect_and_cost() show as None. This means response.completed never fired — check for an interrupted or errored stream, and confirm the event-handling loop isn't exiting early (via an unguarded break) before that event has a chance to arrive.

Common Mistakes

Reading response.id or response.usage directly off the return value of a streaming client.responses.create() call, the way the non-streaming version did — under streaming, this information isn't available until response.completed fires, and code written as if it still worked the old way will fail or silently retrieve stale or missing data.

Not handling a stream interrupted before completion, particularly for the chaining backend, which specifically needs a response ID from the final event to continue correctly — an unhandled early interruption silently breaks the chain for the next turn in a way that can be confusing to debug later.

Forgetting to flush printed output during streaming, reintroducing the batching problem Lesson 2 warned about, and undermining the very responsiveness improvement this project exists to demonstrate.

Best Practices

Reuse a single, shared stream-consumption helper (stream_and_collect() or similar) across all three backends, rather than duplicating event-handling logic in each — this keeps the three implementations consistent and makes a future change to event handling (adding error handling, say) a one-place edit rather than three.

Explicitly test the interruption path, not just the happy path, for at least the chaining backend, since it's the one most affected by an incomplete stream — verify that an interrupted turn doesn't leave the stored chain state in a broken condition for the next call.

Compare the streaming and non-streaming versions of this project side by side, rather than only reading about the difference, to build a concrete, first-hand sense of Lesson 1's perceived-responsiveness argument that no amount of prose description fully substitutes for.

Add the "generating..." indicator and a stop-generating handler as standard features, not optional extras, since both directly reflect genuine user-experience expectations for any real streaming chat interface — a production chat feature without some form of both would feel noticeably less polished than the products most users are already accustomed to.

What This Project Demonstrates, End to End

Having built both the Unit 4 and Unit 5 versions of this chatbot, it's worth stepping back and naming exactly what the comparison between them proves. The Unit 4 version showed that "memory" in a stateless API is entirely a matter of application-level design, with three genuinely different mechanisms achieving a similar user-facing result through different means. This project shows that streaming is an orthogonal concern layered on top of any of those three mechanisms — it changes how content and metadata become available to your code (incrementally, through typed events, rather than all at once on a single object) without changing what memory mechanism is in use underneath. The chaining backend's need to wait for response.completed before it can continue the chain is the clearest illustration of this interaction: it isn't a special case invented for this project, it's the direct, necessary consequence of combining Unit 4, Lesson 3's chaining mechanism with this unit's streaming mechanism, and understanding why that combination behaves the way it does is a genuinely transferable piece of understanding — the same reasoning applies to combining streaming with function calling in Unit 8, or with the Agents SDK in Unit 11, both of which also produce response-level metadata that streaming defers until a completion event fires.

A learner who has built both versions of this project, rather than only read about streaming and memory as separate topics, should come away with a working mental model for a question that will recur throughout the rest of this course: whenever a new capability is introduced, what happens when it's combined with streaming? The answer is very often the same shape seen here — the capability itself works identically, but any of its metadata that isn't part of the incrementally-generated text (an ID, a usage figure, a tool call's arguments) becomes available only once the relevant event in the stream's lifecycle actually fires, typically at or near response.completed, rather than being immediately accessible the way it would be on a plain, synchronous response object.

Suggested Extensions

A few further exercises reinforce this unit's material even more concretely, left for independent practice rather than walked through in full here.

Add background-mode support as a fourth "backend mode" alongside the three memory mechanisms, specifically for a hypothetical --report flag that generates a long, detailed analysis using gpt-6-astra at high reasoning effort submitted via background=True (Lesson 4), polling for completion while the interactive chat loop remains free to handle other commands in the meantime — a small taste of the kind of concurrent job management a more sophisticated application would need.

Instrument all three backends with the token-level chunking inspector from Lesson 2, temporarily, to directly observe how differently-sized delta chunks arrive for the same prompt run several times — a hands-on confirmation that chunk boundaries are not reliably aligned with word boundaries, reinforcing why any content-inspection logic (a keyword-detection feature, say) must operate on accumulated text rather than individual chunks.

Measure and log time-to-first-content for every turn, exactly as Lesson 1 suggested tracking in a production application, and compare the distribution of that metric across the three memory backends — since all three should produce comparable time-to-first-content figures (streaming's benefit is independent of which memory mechanism is in use), this is a good sanity check that the implementation is behaving as this lesson's mental model predicts, and a discrepancy would be worth investigating as a potential bug rather than assumed to be expected variation.