Response Streaming

The previous lesson placed conversation state on the client and wrote the token cost of every send policy. The response of every call arrived there as a single piece: the client sent the request, waited and found a whole text in hand. The printed first token step column had entered no decision.

A response is not actually produced as a single piece. The model builds the output step by step and the interface can send these chunks in the order they are produced. This lesson measures that option and separates three questions: whether streaming changes the total tokens, how much it lowers the time to first token, and what stays usable when the network cuts the response in the middle. The third question is tied directly to the rule of this course: a call whose failure is not written down counts as unmeasured, and a response left half finished is a failure.

• AU9. The code is a simulator; no real endpoint is called. The conversations are fictional, seed 20260218. The simulator carries the network layer set up in the previous lesson.
• AU10. The response is a list of chunks. The simulator also returns the chunks field; the streaming client joins this list in the order it arrives, while the non-streaming client sees it only when the response is complete.
• AU11. The network cut depends on the content of the prompt, not on the field order. Because the three setups send the same prompts, the same calls are cut; the comparison is made under the same network condition.
• AU12. Field order is a request field, not a sentence written into the prompt. So the input tokens stay exactly the same in the three setups and only the order changes.
• AU13. Two separate waits are measured: the time to first token (the moment the user sees something on screen) and the time to actionable output (the moment the client can read the label and start work).
• AU14. The usability criterion is that the label can be read. In the non-streaming client the body of a response cut midway is never in hand; in the streaming client the chunks that arrived stay in hand.
• AU15. The send policy is inherited from the previous lesson and is the same in all three setups: user turns only. The workload is the eight turns of forty conversations, 320 calls in total.
• AU16. Latency is written not as raw time but as steps; the latency class is again the chain.

A Response Arriving Chunk by Chunk

The job of a streaming client is two lines: accumulate the arriving chunk and check whether the accumulated text has become decidable. The streaming_client below does this and writes the step at which it first read the label. whole_client waits for the response to complete; if it does not complete, nothing is left in its hand.

# SIMULATOR -- FICTIONAL. No real endpoint is called; the network layer is simulated too.
SEED, M32 = 20260218, 0xFFFFFFFF
WINDOW, MAX_LENGTH, RATE_EVERY = 240, 24, 37


def rng(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def next_value():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return next_value


def pick(u, w):
    t = 0.0
    for i, x in enumerate(w):
        t += x
        if u < t:
            return i
    return len(w) - 1


def split_pieces(m):
    """FICTIONAL tokenizer; a piece has at most five characters."""
    b = []
    for s in m.lower().replace(",", " , ").split():
        new = True
        while len(s) > 5:
            b.append((s[:5], new))
            s, new = s[5:], False
        b.append((s, new))
    return b


def tokenize(m):
    return [p for p, _ in split_pieces(m)]


def join_pieces(pl):
    s = ""
    for p, new in pl:
        s += (" " if new and s else "") + p
    return s


def hash_text(m):
    h = 2166136261
    for c in m:
        h = ((h ^ ord(c)) * 16777619) & M32
    return h


def draw(prompt, seed, salt):
    return rng((seed + salt) ^ hash_text(prompt))()


TOPICS = ["fatura", "ariza", "sayac", "tarife", "kesinti"]
CONCEPTS = TOPICS + ["odeme", "sikayet", "bilgi", "genel"]
CLUES = {"fatur": "fatura", "tutar": "fatura", "ayrin": "fatura", "borc": "fatura",
         "ariza": "ariza", "bozuk": "ariza", "calis": "ariza", "akmiy": "ariza",
         "sayac": "sayac", "endek": "sayac", "okuma": "sayac", "rande": "sayac",
         "tarif": "tarife", "birim": "tarife", "abone": "tarife", "basam": "tarife",
         "kesin": "kesinti", "duyur": "kesinti", "mahal": "kesinti",
         "odeme": "odeme", "sikay": "sikayet", "bilgi": "bilgi"}
TRAPS = {"degis": "tarife", "kayit": "ariza", "ekip": "kesinti", "takip": "tarife"}
BODY = "talebiniz incelendi ve ilgili ekibe iletildi"
CALL_COUNT = [0]


def flatten(messages):
    return " ".join(f"{r} : {i}" for r, i in messages)


def call(messages, seed=SEED, temperature=0.6, max_length=MAX_LENGTH, position="only"):
    """position is a request field: whether the label is produced at the start or the end."""
    CALL_COUNT[0] += 1
    prompt = flatten(messages)
    toks = tokenize(prompt)
    first = 1 + len(toks) // 40                   # steps until the first token
    empty = {"input": len(toks), "output": 0, "chunks": [], "first": first}
    if len(toks) > WINDOW:
        return dict(empty, status="context_overflow")
    if CALL_COUNT[0] % RATE_EVERY == 0:
        return dict(empty, status="rate_limit")
    r = rng(seed ^ hash_text(prompt))
    if len(toks) > 90 and r() < 0.04:
        return dict(empty, status="timeout")
    candidates = [k for k in CONCEPTS if k[:5] in toks]
    if len(candidates) < 3:
        candidates = list(CONCEPTS)
    p = {k: 0.15 for k in candidates}
    for t in toks:
        for table, w in ((CLUES, 1.0), (TRAPS, 0.6)):
            if table.get(t) in p:
                p[table[t]] += w
    names = sorted(p)
    if temperature <= 0.0:
        k = max(names, key=lambda a: p[a])
    else:
        from math import exp
        e = [exp(p[a] / temperature) for a in names]
        k = names[pick(r(), [x / sum(e) for x in e])]
    if position == "start":
        text = f"konu {k} , {BODY}"
    elif position == "end":
        text = f"{BODY} , konu {k}"
    else:
        text = k
    pl, status = split_pieces(text), "done"
    if len(pl) > max_length:
        pl, status = pl[:max_length], "stopped"
    u = draw(prompt, seed, 91)                    # the network cuts the response midway
    if len(pl) > 3 and u < 0.08:
        pl, status = pl[:1 + int(u / 0.08 * (len(pl) - 1))], "cut"
    return {"status": status, "input": len(toks), "output": len(pl),
            "chunks": pl, "first": first}


def read_label(y):
    return next((s for s in y.split() if s in TOPICS), None)


def whole_client(s):
    """If the response does not complete, the body is never in hand."""
    if s["status"] != "done":
        return "", None
    text = join_pieces(s["chunks"])
    return text, (s["first"] + s["output"]) if read_label(text) else None


def streaming_client(s):
    """Chunks are joined as they arrive; the step is written as soon as the label is read."""
    text, usable_step = "", None
    for i, (p, new) in enumerate(s["chunks"], 1):
        text += (" " if new and text else "") + p
        if usable_step is None and read_label(text):
            usable_step = s["first"] + i
    return text, usable_step


OPENERS = ["merhaba yardim gerekiyor", "iyi gunler destek istiyorum",
           "selam bir sorunum var", "merhaba destek almak istiyorum"]
KEY_TURNS = {"fatura": ["bu ayki fatura tutari gecen aya gore cok yuksek geldi",
                        "faturami odedim ama sistemde borc gorunuyor",
                        "fatura ayrintisini nereden gorebilirim"],
             "ariza": ["evde su akmiyor bir ariza var galiba",
                       "baglanti bozuk oldu ve hic calismiyor",
                       "ariza kaydi actim ama ekip gelmedi"],
             "sayac": ["sayac okumasi yanlis girilmis endeks tutmuyor",
                       "sayac degisimi icin randevu istiyorum",
                       "sayac uzerindeki endeks ile fatura ayni degil"],
             "tarife": ["hangi tarife basamagindayim ogrenmek istiyorum",
                        "tarife degisikligi icin ne yapmam gerekiyor",
                        "birim ucret abone tipine gore degisiyor mu"],
             "kesinti": ["mahallede kesinti var ne zaman gelecek",
                         "planli kesinti duyurusu nereden takip edilir",
                         "aksamdan beri kesinti yasiyoruz"]}
FILLER = ["tesekkur ederim not aldim", "peki bunun icin ne kadar beklemem gerekir",
          "daha once de ayni durum olmustu", "yazili bir bildirim alabilir miyim",
          "bu konuda kayit acildi mi", "degisiklik yapilirsa haber verilir mi",
          "ekip ne zaman gelir", "hafta sonu da ulasabilir miyim"]
SYSTEM = ("sistem", "destek yardimcisisin . etiket kumesi : fatura , ariza , "
                    "sayac , tarife , kesinti .")
QUESTION = "gorusme {n} icin konu etiketi ver"

CONVERSATIONS = []
for i in range(40):                               # 40 FICTIONAL conversations, eight turns each
    r = rng(SEED + 7 * i)
    k = TOPICS[pick(r(), [0.24, 0.22, 0.18, 0.20, 0.16])]
    key_text = KEY_TURNS[k][int(r() * 3)]
    key_pos = 1 + int(r() * 5)
    turns = [OPENERS[int(r() * 4)]]
    for t in range(1, 7):
        turns.append(key_text if t == key_pos else FILLER[int(r() * 8)])
    turns.append(QUESTION.format(n=7301 + i))
    CONVERSATIONS.append({"id": 7301 + i, "topic": k, "key_pos": key_pos + 1, "turns": turns})

g = CONVERSATIONS[0]
messages = [SYSTEM] + [("kullanici", m) for m in g["turns"]]
print(f"conversation {g['id']} , topic {g['topic']} , input"
      f" {call(messages)['input']} tokens")
for position in ("end", "start"):
    s = call(messages, position=position)
    text, usable_step = streaming_client(s)
    print(f"  label {position:<5} output {s['output']:>2}  first token step {s['first']}"
          f"  time to actionable output in streaming client {usable_step}"
          f"  whole response step {whole_client(s)[1]}")
    print(f"             response : {text}")
print("  same request with seed 20260219 ->",
      join_pieces(call(messages, SEED + 1, position="start")["chunks"]))

conversation 7301 , topic kesinti , input 113 tokens
  label end   output 14  first token step 3  time to actionable output in streaming client 17  whole response step 17
             response : talebiniz incelendi ve ilgili ekibe iletildi , konu kesinti
  label start output 14  first token step 3  time to actionable output in streaming client 6  whole response step 17
             response : konu kesinti , talebiniz incelendi ve ilgili ekibe iletildi
  same request with seed 20260219 -> konu kesinti , talebiniz incelendi ve ilgili ekibe iletildi

Same conversation, same prompt and same output length: in both setups the response is 14 chunks. The only thing that changes is where the label stands. When the label is at the end, both the streaming client and the whole client can decide at step 17; streaming gains nothing. When the label is moved to the start, the streaming client decides at step 6, while the whole client again waits for step 17. The eleven steps in between are a wait that could be skipped without reading the body of the response.

The last line follows from the third rule of the course: the same request with a second seed again gave the same label and the same order.

Waiting and Usability Table of the Three Setups

The three setups run on the same 320 calls. The first is non-streaming and asks for the label at the end; the second reads the same request by streaming; the third streams and asks for the label at the start. Because the calls are exactly the same, the network condition is also the same.

def measure(position, streaming, seed=SEED):
    """position: end | start. streaming: whether chunks are read as they arrive."""
    CALL_COUNT[0] = 0
    tok_in = tok_out = cut = cut_recovered = correct = 0
    visible, usable = [], []
    for conv in CONVERSATIONS:
        for t in range(1, 9):
            messages = [SYSTEM] + [("kullanici", m) for m in conv["turns"][:t]]
            s = call(messages, seed, position=position)
            tok_in, tok_out = tok_in + s["input"], tok_out + s["output"]
            if streaming:
                text, step = streaming_client(s)
                if s["chunks"]:
                    visible.append(s["first"] + 1)
            else:
                text, step = whole_client(s)
                if step is not None:
                    visible.append(step)
            if s["status"] == "cut":
                cut += 1
                cut_recovered += step is not None
            if step is not None:
                usable.append(step)
            if t == 8 and read_label(text) == conv["topic"]:
                correct += 1
    avg = sum(visible) / len(visible), sum(usable) / len(usable)
    return tok_in, tok_out, cut, cut_recovered, avg[0], avg[1], len(usable) / 320, correct / 40


SETUPS = [("non-streaming , label end", "end", False),
          ("streaming , label end", "end", True),
          ("streaming , label start", "start", True)]
print("setup                      input output    cut recovered  time to first token"
      "  time to actionable output    usable share  accuracy A  accuracy B")
for name, position, streaming in SETUPS:
    a, b = measure(position, streaming), measure(position, streaming, SEED + 1)
    print(f"{name:<25} {a[0]:>6} {a[1]:>6} {a[2]:>6} {a[3]:>9}  {a[4]:>19.2f}"
          f"  {a[5]:>25.2f}  {a[6]:>14.4f}  {a[7]:>10.4f}  {b[7]:>10.4f}")

CALL_COUNT[0] = 0
share = []
for conv in CONVERSATIONS:
    for t in range(1, 9):
        s = call([SYSTEM] + [("kullanici", m) for m in conv["turns"][:t]], position="start")
        if s["status"] == "cut":
            share.append(s["output"] / 14)
print(f"\nchunk share arriving in {len(share)} cut responses : min {min(share):.4f} ,"
      f" mean {sum(share) / len(share):.4f} , max {max(share):.4f}")

setup                      input output    cut recovered  time to first token  time to actionable output    usable share  accuracy A  accuracy B
non-streaming , label end  23338   4049     33         0                16.03                      16.03          0.8625      0.7500      0.6500
streaming , label end      23338   4049     33         0                 3.29                      16.03          0.8625      0.7500      0.6500
streaming , label start    23338   4049     33        31                 3.29                       4.96          0.9594      0.7750      0.6750

chunk share arriving in 33 cut responses : min 0.0714 , mean 0.5801 , max 0.9286

The first two columns are the most important result of this lesson. Input 23338, output 4049 and the number of cut calls 33 are exactly the same in all three setups. Streaming does not change a single token; it neither lowers nor raises the cost. In a cost accounting table, a streaming and a non-streaming client cannot be told apart.

What changes is the waiting column. In the non-streaming client the time to first token is 16.03 steps on average, because the first thing visible is already the whole response. In the streaming client this drops to 3.29 steps: about one fifth. The appearance of something on the user’s screen is pulled this early, and no token is paid for it.

What Streaming Did Not Gain and Field Order Did

The time to actionable output value of the second row is 16.03, the same as the first row. Streaming brought something to the screen but did not let the client decide, because the field the decision rests on was at the very end of the response. The recovered column of the same row is also zero: none of the 33 cut responses stayed usable. Streaming alone is not a resilience scaffolding.

The third row opens the difference with field order. The time to actionable output falls to 4.96, 31 of the 33 cut responses are recovered, and the usable share rises from 0.8625 to 0.9594. The chunk share arriving in the cut responses is at least 0.0714, on average 0.5801, at most 0.9286: on average more than half of a response cut midway reaches us. Whether that share is of any use depends entirely on which field was produced first.

The last two columns carry a warning. The label accuracy in the three setups is 0.7500 / 0.7500 / 0.7750; on the second seed 0.6500 / 0.6500 / 0.6750. The 0.0250 that field order adds to accuracy, while the difference between the two runs of the same row is 0.1000. This gain counts as unmeasured and is written that way. The measured gain of field order is not in accuracy but in the usability and waiting columns; these are also numbers that do not change from run to run.

Summary

• Streaming does not change the total tokens: all three setups spend 23338 input and 4049 output tokens and are cut in 33 calls.
• Streaming lowers the time to first token from 16.03 to 3.29 and no token is paid for it.
• Streaming alone does not change usability: when the label is at the end, the time to actionable output is 16.03 in both setups and the number recovered from the 33 cut responses is 0.
• When the label is moved to the start of the response, the time to actionable output falls to 4.96, while the number recovered rises to 31 and the usable share to 0.9594.
• On average 0.5801 of the cut responses reaches us; the field order decides whether this share is of any use.
• The 0.0250 that field order adds to accuracy counts as unmeasured because the difference between the two runs is 0.1000; the measured gain is in the waiting and usability columns.

Next Step

In two lessons the response was always a text to be read. When the label came early the work started early, but what started that work was always a person or a simple routing rule. If the response is going to trigger code execution directly, the criterion changes: it is not enough for the format to be readable; the thing called must also exist and the values given must be acceptable. The next lesson ties the model’s output to a call that will be executed and prints three numbers side by side: how many responses produce a call that conforms to the schema, how many fabricate a name that does not exist, and how many of these validation catches.