所有入門單元免費開放 · 課程內容持續更新中

串流:讓回應一個字一個字出現

長回答要等很久才出現,使用者會以為程式當掉。學會串流,讓文字邊產生邊顯示,並把上一堂的聊天程式升級成串流版。

第 10.4 堂約 11 分鐘

上一堂課你做出了會記得前文的聊天程式。不過你可能已經發現一個小困擾:問一個需要長篇回答的問題,畫面會安靜好幾秒,然後整段文字突然冒出來。

對你自己來說還能忍受,但如果是給別人用的工具,使用者很可能以為程式卡住了。聊天網站上那種「一個字一個字打出來」的效果,用的就是這堂課要學的串流(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 之類的參數,再用迴圈逐段讀取,概念相同。

可以這樣問 AI

這是我的串流聊天程式。請幫我在每次回答結束後,多印一行灰色的小字,顯示這次的輸入 token、輸出 token,以及從送出到第一個字出現花了幾秒。

行動練習
  1. 執行最簡單的串流程式,確認文字是一段段出現的。
  2. 把 flush=True 拿掉再執行一次,比較顯示效果有什麼不同。
  3. 把你上一堂的聊天程式升級成串流版,聊三輪確認它仍然記得前文。
  4. 加上 get_final_message(),在每輪結束後印出 stop_reason。

讀完了嗎?

標記完成後,單元與專案的進度會自動更新(只存在這台裝置的瀏覽器)。

卡住了?讓 AI 當你的家教複製這段提問範本,貼到 Claude 或其他 AI 工具,再補上你卡住的地方。

留言與討論

卡關了、有心得,或發現內容需要更新?都歡迎留言。留言使用 GitHub 帳號登入。

留言區即將開放。現在可以先到 GitHub 討論區 發問,或看看 交流頁 的其他管道。