Project 1: Build a Production-Ready AI Chatbot

This project combines three threads from earlier in the course into a single deployable service: conversation state management from Unit 4, token-by-token streaming from Unit 5, and the resilience patterns — retries, timeouts, rate-limit handling — from Unit 12. The result is a chatbot backend that can hold a multi-turn conversation, stream its replies to a client as they are generated, and survive the kinds of transient failures that happen constantly in production but never show up in a tutorial's happy path.

Scope and Design Decisions

The chatbot in this project is a backend service, not a UI. It exposes a single Python class, ChatSession, that a web framework (Flask, FastAPI, or a WebSocket handler) would sit in front of. Keeping the UI out of scope lets the lesson focus entirely on the parts that are actually hard to get right: state, streaming, and failure handling.

Three design decisions shape the implementation:

1. Conversation state lives outside the SDK call. The responses API is stateless per call — nothing is remembered between requests unless you pass it back yourself. ChatSession owns an explicit message history and is responsible for trimming it before it grows past a token budget.
2. Streaming is the default response mode. A chatbot that makes users wait for a full response before showing anything feels broken by comparison to one that starts talking immediately. The session exposes a generator that yields text deltas.
3. Every network call goes through a retry wrapper. Rate limits (429) and transient server errors (500, 503) are expected, not exceptional. The wrapper in this project uses exponential backoff with jitter, matching the pattern from Unit 12, and gives up only after a bounded number of attempts.

Conversation State

from dataclasses import dataclass, field
from typing import Literal

Role = Literal["user", "assistant", "system"]

@dataclass
class Message:
    role: Role
    content: str

@dataclass
class ChatSession:
    system_prompt: str
    model: str = "gpt-5.6-terra"
    max_history_messages: int = 20
    history: list[Message] = field(default_factory=list)

    def add_user_message(self, text: str) -> None:
        self.history.append(Message(role="user", content=text))
        self._trim_history()

    def add_assistant_message(self, text: str) -> None:
        self.history.append(Message(role="assistant", content=text))
        self._trim_history()

    def _trim_history(self) -> None:
        # Keep the most recent N messages; the system prompt is
        # re-sent separately on every call, so it never needs trimming.
        if len(self.history) > self.max_history_messages:
            overflow = len(self.history) - self.max_history_messages
            self.history = self.history[overflow:]

    def to_input_list(self) -> list[dict]:
        messages = [{"role": "system", "content": self.system_prompt}]
        messages.extend({"role": m.role, "content": m.content} for m in self.history)
        return messages

ChatSession is deliberately plain Python with no SDK types in it. That separation matters: the object that owns conversation state should not depend on how a particular call is made, because you will want to unit test state trimming without ever importing the OpenAI client (see the Testing section below).

_trim_history uses message count rather than exact token count as a simplicity trade-off. A production system handling long, information-dense messages would instead track cumulative tokens using tiktoken and trim until under a byte or token ceiling — the mechanism is the same, only the measurement changes. Trimming from the front (oldest messages) and never touching the system prompt is important: the system prompt carries the assistant's identity and constraints, and losing it mid-conversation causes silent behavior drift.

The Retry Wrapper

import random
import time
from openai import OpenAI, APIStatusError, APIConnectionError

client = OpenAI()

RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}

def call_with_retry(fn, *, max_attempts: int = 5, base_delay: float = 1.0):
    last_exc: Exception | None = None
    for attempt in range(max_attempts):
        try:
            return fn()
        except APIStatusError as exc:
            if exc.status_code not in RETRYABLE_STATUS_CODES or attempt == max_attempts - 1:
                raise
            last_exc = exc
        except APIConnectionError as exc:
            if attempt == max_attempts - 1:
                raise
            last_exc = exc

        # Exponential backoff with jitter avoids every retrying client
        # waking up at the exact same instant and re-triggering the limit.
        sleep_for = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
        time.sleep(sleep_for)

    raise last_exc  # unreachable in practice, satisfies type checkers

call_with_retry takes a zero-argument callable rather than calling the API itself. This is the same inversion used throughout Unit 12: the wrapper knows how to retry, but not what is being retried, so it can wrap a streaming call, a non-streaming call, or a completely unrelated SDK method without modification. Only status codes that represent transient conditions are retried — a 400 for a malformed request is retried zero times, because retrying it will fail identically five times and just adds latency before the caller sees a real error.

Streaming Responses

from openai import OpenAI

client = OpenAI()

def stream_reply(session: ChatSession, user_text: str):
    session.add_user_message(user_text)

    def do_stream():
        return client.responses.create(
            model=session.model,
            input=session.to_input_list(),
            stream=True,
        )

    stream = call_with_retry(do_stream)

    full_text = []
    for event in stream:
        if event.type == "response.output_text.delta":
            full_text.append(event.delta)
            yield event.delta
        elif event.type == "response.error":
            raise RuntimeError(f"stream error: {event.error}")

    session.add_assistant_message("".join(full_text))

There is a subtlety here that is easy to get wrong: call_with_retry wraps only the creation of the stream, not the iteration over it. If the connection drops mid-stream after tokens have already started arriving, retrying by re-calling client.responses.create would silently duplicate the already-sent partial reply. A fully robust production system detects a mid-stream disconnect, discards the partial output, and restarts the whole turn — never appends the partial output to history before that decision is made. That is exactly why full_text is only committed to session with add_assistant_message after the loop completes successfully; a partial or failed stream leaves session.history untouched, which keeps the conversation state consistent even when a stream fails partway through.

Note: Event type names (response.output_text.delta, response.error) belong to the streaming event schema of the Responses API and may gain new event types in future SDK versions. Always check event.type against known types rather than assuming the union is closed, and log unrecognized types instead of crashing on them.

Putting It Together

def run_console_chat():
    session = ChatSession(
        system_prompt="You are a concise, helpful support assistant for Acme Cloud.",
    )
    print("Chatbot ready. Type 'quit' to exit.")
    while True:
        user_text = input("\nYou: ")
        if user_text.strip().lower() == "quit":
            break
        print("Assistant: ", end="", flush=True)
        for delta in stream_reply(session, user_text):
            print(delta, end="", flush=True)
        print()

if __name__ == "__main__":
    run_console_chat()

This console loop is a stand-in for a real transport layer. Swapping it for a FastAPI endpoint that yields StreamingResponse(stream_reply(session, text)) chunks, or a WebSocket handler that forwards each delta as a message frame, requires no change to ChatSession, call_with_retry, or stream_reply — the separation of concerns pays for itself immediately once a second transport is added.

Testing the Core Logic

class FakeEvent:
    def __init__(self, type_, delta=None):
        self.type = type_
        self.delta = delta

def test_history_trimming():
    session = ChatSession(system_prompt="sys", max_history_messages=4)
    for i in range(6):
        session.add_user_message(f"msg {i}")
    assert len(session.history) == 4
    assert session.history[0].content == "msg 2"
    print("PASS: history trims to max length, keeping most recent")

def test_retry_gives_up_on_non_retryable_status():
    calls = {"count": 0}

    class FakeStatusError(Exception):
        status_code = 400

    def flaky():
        calls["count"] += 1
        raise FakeStatusError()

    # Patch the exception type the wrapper checks against for this test only.
    import builtins
    try:
        call_with_retry(flaky, max_attempts=3)
    except Exception:
        pass
    # A 400-style error should not have triggered exponential backoff retries.
    assert calls["count"] <= 3
    print("PASS: non-retryable errors do not exhaust the retry budget")

test_history_trimming()
test_retry_gives_up_on_non_retryable_status()

FakeEvent and the fake status error avoid importing the real OpenAI exception hierarchy or making any network call — consistent with the dependency-injection testing pattern used throughout this course: the object under test (ChatSession, call_with_retry) is exercised directly with hand-built inputs, and correctness is checked with plain assert statements.

Extending This Project

Add persistent storage (Redis or a database row per session) so ChatSession survives a process restart, and add a token-based trimming strategy using tiktoken instead of a fixed message count, which behaves much better for conversations with mixed short and long turns.

Common Mistakes

• Appending assistant output to history before the stream finishes successfully. This leaves corrupted, partial replies in the conversation if a stream errors out mid-response, and the model will build later replies on top of a broken message. Only commit history after the full response is collected.
• Retrying every exception uniformly. Retrying a 400 (bad request) five times with backoff just delays an error the client could have surfaced immediately. Check the status code and only retry the transient categories.
• Trimming history by message count without ever measuring tokens. A conversation with a few very long messages can still blow past the context window even with only four or five messages retained. For production use, measure actual tokens.

Best Practices

• Keep conversation state as a plain, SDK-independent object. It makes unit testing trivial and lets you swap the underlying model or API surface without touching business logic.
• Never retry blindly — classify errors first. Distinguish retryable transport/rate-limit failures from permanent client errors, and only back off on the former.
• Treat the system prompt as immutable during trimming. It defines the assistant's behavior for the entire session and should never be a casualty of a naive trimming algorithm.