串流:讓回應一個字一個字出現
長回答要等很久才出現,使用者會以為程式當掉。學會串流,讓文字邊產生邊顯示,並把上一堂的聊天程式升級成串流版。
上一堂課你做出了會記得前文的聊天程式。不過你可能已經發現一個小困擾:問一個需要長篇回答的問題,畫面會安靜好幾秒,然後整段文字突然冒出來。
對你自己來說還能忍受,但如果是給別人用的工具,使用者很可能以為程式卡住了。聊天網站上那種「一個字一個字打出來」的效果,用的就是這堂課要學的串流(streaming)。
串流在做什麼
模型產生文字本來就是一小段一小段接著寫的。一般呼叫會等它全部寫完,才把整份回應交給你;串流則是寫好一段就先送一段過來。
串流不會讓總時間變短,改變的是「等多久才看到第一個字」。
最簡單的串流程式
from dotenv import load_dotenv
import anthropic
load_dotenv()
MODEL = "claude-sonnet-5-5" # 可以換成其他模型 ID
client = anthropic.Anthropic()
with client.messages.stream(
model=MODEL,
max_tokens=64000,
messages=[{"role": "user", "content": "寫一篇 300 字的短文,主題是第一次學寫程式的心情。"}],
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
print() # 最後換行
跟上一堂相比只改了兩個地方:
messages.create換成messages.stream,並用with包起來。with會在結束時自動幫你關閉連線。- 用
for text in stream.text_stream逐段拿到文字。
print 的兩個參數也很重要:end="" 讓每段文字接在同一行,不要自動換行;flush=True 讓文字立刻顯示,不要先存在緩衝區裡。少了 flush=True,你可能會看到文字還是一大塊一大塊出現。
串流結束後,拿回完整的回應
串流時你只拿到一段段文字,那用量、停止原因去哪裡看?在 with 區塊裡呼叫 get_final_message(),就能拿到和一般呼叫一模一樣的完整回應物件:
with client.messages.stream(
model=MODEL,
max_tokens=64000,
messages=[{"role": "user", "content": "用五點說明怎麼開始學 Python。"}],
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
final = stream.get_final_message()
print()
print("停止原因:", final.stop_reason)
print("輸出 token:", final.usage.output_tokens)
把聊天程式升級成串流版
把上一堂的聊天迴圈改成串流,只要換掉呼叫模型的那一段:
from dotenv import load_dotenv
import anthropic
load_dotenv()
MODEL = "claude-sonnet-5-5"
client = anthropic.Anthropic()
SYSTEM = "你是一位親切的學習夥伴,用繁體中文回答。"
history = []
print("開始聊天,輸入 exit 結束。")
while True:
user_text = input("\n你:").strip()
if user_text.lower() == "exit":
break
if not user_text:
continue
history.append({"role": "user", "content": user_text})
print("AI:", end="", flush=True)
with client.messages.stream(
model=MODEL,
max_tokens=64000,
system=SYSTEM,
messages=history,
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
final = stream.get_final_message()
print()
# 跟上一堂一樣,把完整內容存回歷史
history.append({"role": "assistant", "content": final.content})
存回歷史的寫法和上一堂完全一樣,只是 response 換成了 final。
什麼時候該用串流
| 情境 | 建議 |
|---|---|
| 顯示給人看的聊天、寫作助手 | 用串流,體驗好很多 |
回答可能很長,max_tokens 設得很大 |
用串流,也能避免長時間等待造成連線逾時 |
| 背景批次處理、只要結果存進檔案 | 一般呼叫就好,程式比較簡單 |
範例裡的 max_tokens 設成 64000,因為串流比較不怕長回答造成逾時,可以放心給模型足夠的空間。
其他供應商的 SDK 也都有串流功能,通常是在呼叫時加上 stream=True 之類的參數,再用迴圈逐段讀取,概念相同。
這是我的串流聊天程式。請幫我在每次回答結束後,多印一行灰色的小字,顯示這次的輸入 token、輸出 token,以及從送出到第一個字出現花了幾秒。
- 執行最簡單的串流程式,確認文字是一段段出現的。
- 把
flush=True拿掉再執行一次,比較顯示效果有什麼不同。 - 把你上一堂的聊天程式升級成串流版,聊三輪確認它仍然記得前文。
- 加上
get_final_message(),在每輪結束後印出stop_reason。
讀完了嗎?
標記完成後,單元與專案的進度會自動更新(只存在這台裝置的瀏覽器)。
留言與討論
卡關了、有心得,或發現內容需要更新?都歡迎留言。留言使用 GitHub 帳號登入。