Multi-turn 與 streaming

TL;DR

• Anthropic API 是 stateless——它不記得你上一輪說了什麼，整段歷史每次都要重傳
• assistant 回應後，把它塞回 messages array 才能延續對話
• Streaming 讓 user 在 5–30 秒的等待期間看到 token 流出來；tool use 場景幾乎是必備

一個情境：為什麼越聊越貴

寫第一個 chat app 的時候很容易嚇到——前 3 句對話 token 數還好，到第 20 句一次 request 5000+ token 起跳，帳單上升超有感。

原因是 API 不記憶。你每次發 request 都要把過去所有對話一起傳。20 句 = 你連續送了 1+2+3+...+20 = 210 條訊息給 model 處理。

這不是 bug 是設計：stateless 讓 API 可以負載平衡到任意機器，你也可以隨時編輯歷史（刪掉某些 turn、改寫某些回答）。但代價是對話狀態管理是你的工作。

Multi-turn 的正確做法

messages = []

# 第一輪
messages.append({"role": "user", "content": "什麼是量子運算？一句話。"})
res = client.messages.create(
    model="claude-sonnet-4-6", max_tokens=1024, messages=messages
)
assistant_text = res.content[0].text

# 把 assistant 回應塞回去！
messages.append({"role": "assistant", "content": assistant_text})

# 第二輪——Claude 看得到完整對話
messages.append({"role": "user", "content": "再寫一句更詳細的"})
res = client.messages.create(
    model="claude-sonnet-4-6", max_tokens=1024, messages=messages
)

漏掉那行 messages.append({"role": "assistant", ...}) 是新手最常見的錯。沒塞回去 Claude 就以為「再寫一句更詳細的」是憑空冒出來的，會回不知所云。

包成 helper

def add_user(messages, text):
    messages.append({"role": "user", "content": text})

def add_assistant(messages, text):
    messages.append({"role": "assistant", "content": text})

之後都用：

messages = []
add_user(messages, "什麼是量子運算？")
res = chat(messages)
add_assistant(messages, res.content[0].text)
add_user(messages, "再寫一句")
res = chat(messages)

Token 累積是真議題

每次都重傳完整歷史 = token 數線性成長。20 turn 的對話可能單次 request 就 5000+ input tokens。應對策略三選一：

策略	適用
設 turn 數上限（例如最多保留最近 10 輪）	chat app、不需要長期記憶
背景做 summary（每 N 輪叫 model 摘要前面，新對話帶 summary 不帶原文）	agent、長 session
prompt caching（system + 前面 turn cache 起來，付便宜的讀取錢）	一份長 doc 反覆問答

第三個 後面章節 會講。

Streaming 是什麼

不開 stream 的時候，你 send 一個 request 就 block 5–30 秒，等 model 跑完才一次拿到完整 response。User 介面只能轉 spinner。

開 stream 之後，model 生一個 token 就回一個，你可以邊收邊送給前端，user 看到打字機效果。

streaming 也不只是 UX。它讓你可以早一點知道發生什麼：

• 第一個 token 通常數百 ms 到 1–2 秒到（依 model 與長度），TTFT（time-to-first-token）是 chat app 的關鍵指標
• 看到特定 token 出現可以提早觸發後續動作（例如看到 <tool_use> 開頭就準備執行環境）
• 如果 model 跑歪了可以提早 cancel 省 token

兩種 stream 寫法

1. 原始 event 流（看細節用）

stream = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=messages,
    stream=True,
)

for event in stream:
    print(event.type, event)

你會看到一串 event：

event	意思
message_start	回應開始
content_block_start	一個 block（text / tool_use / thinking）開始
content_block_delta	block 裡面新加的內容（真正的 token 在這）
content_block_stop	一個 block 結束
message_delta	message 級別的更新（usage、stop_reason）
message_stop	整個 response 結束

2. SDK 簡化版（chat app 直接用）

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=messages,
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()  # 完整 message 拿來存 DB

stream.text_stream 已經幫你過濾出 text delta，chat app 99% 用這個就好。最後 get_final_message() 給你完整 message 物件去存 DB。

什麼時候不該開 stream

• 批次處理：CI 上跑 1000 個分類，沒人在看打字機，要的是吞吐量
• eval pipeline：拿到完整 response 才能打分
• structured JSON 輸出：JSON 沒生完是壞的，stream 反而麻煩

接下來

下一篇處理兩個常見痛點：怎麼用 temperature 控制隨機性、怎麼穩拿到 JSON / 結構化輸出（從 prefill + stop_sequence 開始；後面 tool use 章節有更穩的解法）。