Streaming Responses

# script_id: day_013_streaming/blocking_baseline
# blocking_baseline.py: time one long answer without streaming.
# Install: uv add "openai>=3.46,<4"   Env: OPENAI_API_KEY (ATLAS_MODEL optional)
import os
import time

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()

QUESTION = ("Explain how job leases and heartbeats work in a background-jobs library, "
            "in about 300 words.")
t0 = time.perf_counter()
resp = client.responses.create(model=MODEL, input=QUESTION, reasoning={"effort": "none"})
total = time.perf_counter() - t0   # the screen stays blank for this entire interval
print(resp.output_text)
print(f"\ntotal {total:.1f} s · {resp.usage.output_tokens} output tokens")

Imagine a job queue where a worker takes a job off the queue...   (300 words appear at once)

total 4.8 s · 412 output tokens        (your numbers will differ)

# script_id: day_013_streaming/fold_by_hand
# fold_by_hand.py: fold a recorded stream into text. No API key or install needed.
# Hand-written to match the SDK's event types; real events carry more fields (ids, indexes).
EVENTS = [
    {"type": "response.created", "sequence_number": 0},
    {"type": "response.output_item.added", "sequence_number": 1, "item": {"type": "message"}},
    {"type": "response.output_text.delta", "sequence_number": 2, "delta": "Leases"},
    {"type": "response.output_text.delta", "sequence_number": 3, "delta": " expire"},
    {"type": "response.output_text.delta", "sequence_number": 4, "delta": " without heartbeats."},
    {"type": "response.completed", "sequence_number": 5,
     "response": {"status": "completed", "usage": {"output_tokens": 5}}},
]

text = ""
for ev in EVENTS:
    if ev["type"] == "response.output_text.delta":
        text += ev["delta"]            # a delta is only the NEW piece; append it
        print(f"after #{ev['sequence_number']}: {text!r}")
    elif ev["type"] == "response.completed":
        print("final:", text, "| status:", ev["response"]["status"],
              "| usage:", ev["response"]["usage"])

after #2: 'Leases'
after #3: 'Leases expire'
after #4: 'Leases expire without heartbeats.'
final: Leases expire without heartbeats. | status: completed | usage: {'output_tokens': 5}

# script_id: day_013_streaming/event_dump
# event_dump.py: list the event types one streamed call produces, in order of first appearance.
# Install: uv add "openai>=3.46,<4"   Env: OPENAI_API_KEY (ATLAS_MODEL optional)
import os

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()

seen: set[str] = set()
with client.responses.create(model=MODEL, input="In two sentences, what does `orbit doctor` check?",
                             reasoning={"effort": "none"}, stream=True) as stream:
    for ev in stream:
        if ev.type not in seen:
            seen.add(ev.type)
            print(f"#{ev.sequence_number:<4} {ev.type}")

#0    response.created
#1    response.in_progress
#2    response.output_item.added
#3    response.content_part.added
#4    response.output_text.delta
#41   response.output_text.done
#42   response.content_part.done
#43   response.output_item.done
#44   response.completed          (typical; you may also see a reasoning item, and the numbers differ)

# script_id: day_013_streaming/minimal_stream
# minimal_stream.py: print an answer while it is generated.
# Install: uv add "openai>=3.46,<4"   Env: OPENAI_API_KEY (ATLAS_MODEL optional)
import os

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()

stream = client.responses.create(
    model=MODEL,
    input="Explain how job leases and heartbeats work in a background-jobs library, in about 300 words.",
    reasoning={"effort": "none"},   # no hidden thinking before the first word
    stream=True,                    # returns an iterator of typed events, not a Response
)
for ev in stream:
    if ev.type == "response.output_text.delta":
        print(ev.delta, end="", flush=True)   # flush, or the terminal may hold the text back
print()

Imagine a job queue where a worker takes a job off the queue... (words appear a few at a time)

# script_id: day_013_streaming/break_it
# fragment: three edits to minimal_stream.py. Try each one on its own.

# 1. The Chat Completions habit. Responses events have no .choices:
for chunk in stream:
    print(chunk.choices[0].delta.content)
# AttributeError: 'ResponseCreatedEvent' object has no attribute 'choices'

# 2. Drop flush=True. Text shows up in bursts, or only at the end when output is piped:
    print(ev.delta, end="")

# 3. Raise the effort. The model reasons privately before it writes the first word:
stream = client.responses.create(model=MODEL, input=QUESTION, stream=True,
                                 reasoning={"effort": "medium"})

effort none:   TTFT 0.4 s · total 4.6 s
effort medium: TTFT 3.1 s · total 7.9 s      (illustrative; your numbers will differ)

# script_id: day_013_streaming/stream_complete
# stream_complete.py: Atlas's streaming wrapper (adds to atlas/llm.py). Built on Day 12's llm.py.
# Install: uv add "openai>=3.46,<4" httpx2   Env: OPENAI_API_KEY (ATLAS_MODEL, ATLAS_LOG_DIR optional)
from __future__ import annotations

import json
import os
import time
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import httpx2   # installed with openai 3.x; the SDK's HTTP layer
import openai
from openai import OpenAI
from openai.types.responses import Response

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
EFFORT = "none"   # chat replies: lowest TTFT. Pass reasoning=... to override per call
LOG_DIR = Path(os.environ.get("ATLAS_LOG_DIR", "logs"))
client = OpenAI()
END_EVENTS = {"response.completed", "response.incomplete", "response.failed"}


@dataclass
class StreamStats:
    text: str                      # everything the user saw
    ttft_s: float | None           # None if no text ever arrived
    total_s: float
    output_tokens: int | None      # None when cancelled: usage only comes in the final event
    tokens_per_s: float | None
    cost_usd: float | None
    response: Response | None      # the final Response, or None if you stopped first
    interrupted: bool = False
    error: str | None = None


def print_delta(delta: str) -> None:
    print(delta, end="", flush=True)


def visible_rate(resp: Response, ttft_s: float | None, total_s: float) -> float | None:
    """Visible tokens per second after the first token. Reasoning tokens are never streamed."""
    u = resp.usage
    if u is None or ttft_s is None or total_s <= ttft_s:
        return None
    visible = u.output_tokens - u.output_tokens_details.reasoning_tokens
    return visible / (total_s - ttft_s)

# script_id: day_013_streaming/helper_collect
# helper_collect.py: stream with the SDK helper, then read the final Response and its usage.
# Install: uv add "openai>=3.46,<4"   Env: OPENAI_API_KEY (ATLAS_MODEL optional)
import os

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()

parts: list[str] = []
with client.responses.stream(model=MODEL, input="In three sentences, why do job leases expire?",
                             reasoning={"effort": "none"}) as stream:
    for ev in stream:
        if ev.type == "response.output_text.delta":
            parts.append(ev.delta)                 # keep what the user saw
            print(ev.delta, end="", flush=True)
    final = stream.get_final_response()            # needs a response.completed event

collected = "".join(parts)
print(f"\nmatch: {collected == final.output_text} · status: {final.status} · "
      f"input {final.usage.input_tokens} · output {final.usage.output_tokens} tokens")

A lease is a time limit on a worker's claim to a job...
match: True · status: completed · input 19 · output 74 tokens      (your numbers will differ)

# script_id: day_013_streaming/stream_complete
# Compact copies of Day 12's cost tracker and call log (see ../Day_012_openai_sdk_production/).
# USD per 1M tokens, as of 2026-11, verify at https://developers.openai.com/api/docs/pricing
PRICES_PER_1M = {"gpt-6-luna": {"input": 0.10, "cached_input": 0.01, "output": 0.50}}


def cost_from_usage(usage, model: str = MODEL) -> float:
    p = PRICES_PER_1M.get(model) or next(v for k, v in PRICES_PER_1M.items() if model.startswith(k))
    cached = usage.input_tokens_details.cached_tokens
    return ((usage.input_tokens - cached) * p["input"] + cached * p["cached_input"]
            + usage.output_tokens * p["output"]) / 1_000_000


class CostTracker:
    def __init__(self) -> None:
        self.calls: list[dict] = []

    @property
    def total_usd(self) -> float:
        return sum(c["cost_usd"] for c in self.calls)

    def record(self, response: Response, *, label: str = "", latency_s: float | None = None) -> float:
        cost = cost_from_usage(response.usage, response.model)
        self.calls.append({"label": label, "cost_usd": cost, "latency_s": latency_s})
        return cost

    def summary(self) -> str:
        return f"{len(self.calls)} calls · ${self.total_usd:.5f}"


tracker = CostTracker()


def log_call(response: Response | None, *, label: str, latency_s: float,
             cost_usd: float | None, extra: dict | None = None) -> None:
    """One JSON line per call. Unlike Day 12's copy, accepts None for a cancelled stream."""
    LOG_DIR.mkdir(parents=True, exist_ok=True)
    row = {"ts": time.time(), "label": label, "id": getattr(response, "id", None),
           "status": getattr(response, "status", "cancelled"), "latency_s": round(latency_s, 3),
           "cost_usd": cost_usd, **(extra or {})}
    with (LOG_DIR / "calls.jsonl").open("a", encoding="utf-8") as f:
        f.write(json.dumps(row) + "\n")

# script_id: day_013_streaming/stream_complete
def stream_complete(input: str | list, *, instructions: str | None = None, model: str = MODEL,
                    on_delta: Callable[[str], None] = print_delta, max_seconds: float | None = None,
                    label: str = "stream", **kwargs: Any) -> StreamStats:
    """Stream one Responses call. Ctrl-C, max_seconds or a mid-stream error stop it early."""
    kwargs.setdefault("reasoning", {"effort": EFFORT})
    if instructions is not None:
        kwargs["instructions"] = instructions   # re-sent every call, as on Day 8
    parts: list[str] = []
    final: Response | None = None
    ttft: float | None = None
    interrupted, error, t0 = False, None, time.perf_counter()
    # Errors before the first event (auth, 400, 429 after retries) raise here, as on Day 12.
    stream = client.responses.create(model=model, input=input, stream=True, **kwargs)
    with stream:
        try:
            for ev in stream:
                if ev.type == "response.output_text.delta":
                    ttft = ttft if ttft is not None else time.perf_counter() - t0
                    parts.append(ev.delta)
                    on_delta(ev.delta)
                elif ev.type in END_EVENTS:
                    final = ev.response            # status + usage for all three end states
                elif ev.type == "error":
                    interrupted, error = True, f"stream error {ev.code}: {ev.message}"
                    break
                if max_seconds is not None and final is None and time.perf_counter() - t0 > max_seconds:
                    interrupted = True
                    break                          # leaving `with` closes the connection
        except KeyboardInterrupt:
            interrupted = True
        except (openai.APIError, httpx2.TransportError) as e:   # mid-stream: never auto-retried
            interrupted, error = True, f"{type(e).__name__}: {e}"
    total = time.perf_counter() - t0
    cost = tokens = rate = None
    if final is not None and final.usage is not None:
        cost = tracker.record(final, label=label, latency_s=total)
        tokens, rate = final.usage.output_tokens, visible_rate(final, ttft, total)
    log_call(final, label=label, latency_s=total, cost_usd=cost,
             extra={"ttft_s": ttft, "interrupted": interrupted, "error": error})
    return StreamStats("".join(parts), ttft, total, tokens, rate, cost, final, interrupted, error)

# script_id: day_013_streaming/stream_complete
def footer(s: StreamStats) -> str:
    """One status line per call, e.g. 'TTFT 0.42 s · total 3.9 s · 58 tok/s · $0.00011'."""
    bits = [f"TTFT {s.ttft_s:.2f} s" if s.ttft_s is not None else "no text", f"total {s.total_s:.1f} s"]
    if s.tokens_per_s is not None:
        bits.append(f"{s.tokens_per_s:.0f} tok/s")
    bits.append(f"${s.cost_usd:.5f}" if s.cost_usd is not None else "cost unknown (no usage)")
    if s.interrupted:
        bits.append("[interrupted]")
    if s.error:
        bits.append(s.error)
    return " · ".join(bits)


if __name__ == "__main__":
    q = "Explain how job leases and heartbeats work in a background-jobs library, in about 300 words."
    print("\n" + footer(stream_complete(q, label="leases_full")))
    print("\n" + footer(stream_complete(q, max_seconds=1.5, label="leases_deadline")))
    print(tracker.summary())

Imagine a job queue where a worker takes a job off the queue... (full answer streams)
TTFT 0.41 s · total 3.9 s · 108 tok/s · $0.00022
Imagine a job queue where a worker takes a job off the queue, and it promises
TTFT 0.39 s · total 1.5 s · cost unknown (no usage) · [interrupted]
1 calls · $0.00022                                  (your numbers will differ)

# script_id: day_013_streaming/stream_chat
# stream_chat.py: streaming Atlas chat over Orbit's FAQ, with a footer per turn.
# Install: uv add "openai>=3.46,<4" httpx2   Env: OPENAI_API_KEY; ORBIT_DATA (default datasets/orbit)
# Needs stream_complete.py (above) in the same folder; it calls client.responses.create.
import os
from pathlib import Path

from stream_complete import StreamStats, footer, stream_complete, tracker

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
ATLAS_PERSONA = """You are Atlas, an assistant for the maintainers of Orbit, an open-source Python background-jobs library.
Rules:
- Use only the facts between the <facts> tags. If the answer is not there, say you don't know.
- Never invent config keys, versions or error codes.
Format: at most 4 sentences or 4 bullets."""   # compact version of Day 8's persona


def load_facts(path: Path, max_chars: int = 2400) -> str:
    """Day 8's helper: a docs page without its YAML front matter, cut at a paragraph."""
    text = path.read_text(encoding="utf-8")
    if text.startswith("---"):
        text = text.split("---", 2)[2]
    text = text.strip()
    cut = text.rfind("\n\n", 0, max_chars) if len(text) > max_chars else len(text)
    return text[: cut if cut > 0 else max_chars]


INSTRUCTIONS = f"{ATLAS_PERSONA}\n\n<facts>\n{load_facts(DATA_DIR / 'docs' / 'faq.md')}\n</facts>"

# script_id: day_013_streaming/stream_chat
def chat_turn_streamed(history: list, user_text: str) -> tuple[list, StreamStats]:
    turn_input = [*history, {"role": "user", "content": user_text}]
    stats = stream_complete(turn_input, instructions=INSTRUCTIONS, label="chat")
    if stats.response is not None and not stats.interrupted:
        return [*turn_input, *stats.response.output], stats       # same as Day 8
    # Cancelled or broken: keep exactly what the user saw, once, and say so.
    partial = {"role": "assistant", "content": f"{stats.text} [interrupted]"}
    return [*turn_input, partial], stats


def main() -> None:
    history: list = []
    print("Atlas (streaming). Ctrl-C stops an answer; Ctrl-D or 'quit' exits.")
    while True:
        try:
            user_text = input("\nyou> ").strip()
        except (EOFError, KeyboardInterrupt):
            break
        if user_text.lower() in {"quit", "exit"}:
            break
        if not user_text:
            continue
        print("atlas> ", end="", flush=True)
        try:
            history, stats = chat_turn_streamed(history, user_text)
        except KeyboardInterrupt:          # pressed before the first event arrived
            print("\n  [cancelled before the stream started; history unchanged]")
            continue
        print(f"\n  [{footer(stats)}]")
    print(f"\n{tracker.summary()}")


if __name__ == "__main__":
    main()

Atlas (streaming). Ctrl-C stops an answer; Ctrl-D or 'quit' exits.

you> Does Orbit run each job exactly once?
atlas> No. Orbit delivers jobs at least once. A job whose lease expires (ORB-1101) or whose
worker dies can run again, so make side effects idempotent.
  [TTFT 0.42 s · total 1.3 s · 61 tok/s · $0.00012]

you> Is Windows OK for production?
atlas> For development, yes. For production, the docs recommend against it, because the Windows
worker ignores ^C
  [TTFT 0.40 s · total 0.9 s · cost unknown (no usage) · [interrupted]]

you> quit

1 calls · $0.00012                                  (your numbers will differ)