Streaming 與多模態輸入

Step 1:最小串流範例(Python)

建立 stream_basic.py:

import anthropic
from dotenv import load_dotenv

load_dotenv()
client = anthropic.Anthropic()

print("Claude 說:", end=" ", flush=True)

with client.messages.stream(
    model="claude-haiku-4-5",
    max_tokens=512,
    messages=[{
        "role": "user",
        "content": "用三個重點解釋什麼是 Streaming API,每點一句話。"
    }]
) as stream:
    for text in stream.text_stream:   # text_stream 直接給你文字片段
        print(text, end="", flush=True)

print()  # 最後換行

# 取最終完整的 message 物件(stream 結束後才能拿)
final_msg = stream.get_final_message()
print(f"\n輸入 token: {final_msg.usage.input_tokens}")
print(f"輸出 token: {final_msg.usage.output_tokens}")

幾個重點:

• client.messages.stream() 用 with context manager 包起來,結束後自動關閉連線
• stream.text_stream 是一個 generator,每次迭代給一小段文字(可能只有一兩個字)
• print(text, end="", flush=True) 的 end="" 不換行、flush=True 立刻推到 terminal,不然 Python 的 output buffer 會讓你感覺不到串流效果
• stream.get_final_message() 要在 with 區塊結束後才呼叫,拿到完整的 Message 物件

執行:

python stream_basic.py

你應該會看到文字逐字出現在 terminal,而不是等一下子跳出一大段。

Step 2:串流範例(JavaScript)

建立 stream_basic.mjs:

import Anthropic from "@anthropic-ai/sdk";
import "dotenv/config";

const client = new Anthropic();

process.stdout.write("Claude 說: ");

// 高階寫法:用 stream() + 事件監聽
const stream = client.messages.stream({
    model: "claude-haiku-4-5",
    max_tokens: 512,
    messages: [{
        role: "user",
        content: "用三個重點解釋什麼是 Streaming API,每點一句話。"
    }]
});

// 每有新文字片段就觸發
stream.on("text", (text) => {
    process.stdout.write(text);  // 不換行、不加 buffer
});

// 等 stream 全部完成,取得 final message
const finalMessage = await stream.finalMessage();
console.log("\n");
console.log(`輸入 token: ${finalMessage.usage.input_tokens}`);
console.log(`輸出 token: ${finalMessage.usage.output_tokens}`);

執行:

node stream_basic.mjs

JS SDK 的 .on("text", handler) 是事件驅動式;Python SDK 的 for text in stream.text_stream 是迭代式。兩種風格都是官方支援的寫法,效果相同。

Step 3:圖片輸入——base64 vs URL

圖片有兩種傳法,各有適合場景:

支援的圖片格式:JPEG、PNG、GIF、WebP。不支援 SVG、BMP 或 HEIC。

base64 方式:

import base64
import anthropic
from dotenv import load_dotenv

load_dotenv()
client = anthropic.Anthropic()

# 讀取本機圖片並轉 base64
with open("receipt.jpg", "rb") as f:
    image_data = base64.standard_b64encode(f.read()).decode("utf-8")

response = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=512,
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": "image/jpeg",   # 要和實際格式一致
                    "data": image_data,
                }
            },
            {
                "type": "text",
                "text": "請列出這張收據的店名、日期和總金額。"
            }
        ]
    }]
)
print(response.content[0].text)

URL 方式:

response = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=512,
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "image",
                "source": {
                    "type": "url",
                    "url": "https://example.com/receipt.jpg",  # 必須公開可存取
                }
            },
            {
                "type": "text",
                "text": "請列出這張收據的店名、日期和總金額。"
            }
        ]
    }]
)
print(response.content[0].text)

多張圖片同時傳就在 content 陣列裡加多個 image block,最多可以傳 100 張。不過 token 消耗會等比增加——每張圖大約消耗 1,000–4,000 token,視解析度而定(高解析度圖片會被自動縮到 1568px 以內再計算 token)。

Step 4:PDF 文件輸入

PDF 用 document 類型的 block,目前只支援 base64 方式(不支援 URL 直傳 PDF)。Claude 4 系列的所有模型都支援 PDF 輸入。

import base64
import anthropic
from dotenv import load_dotenv

load_dotenv()
client = anthropic.Anthropic()

# 讀 PDF 並轉 base64
with open("contract.pdf", "rb") as f:
    pdf_data = base64.standard_b64encode(f.read()).decode("utf-8")

# PDF 分析建議用 Sonnet,複雜文件用 Opus
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "document",
                "source": {
                    "type": "base64",
                    "media_type": "application/pdf",  # 固定這個值
                    "data": pdf_data,
                }
            },
            {
                "type": "text",
                "text": "請摘要這份文件的重點,條列式,不超過十點。"
            }
        ]
    }]
)
print(response.content[0].text)

**PDF 的 token 計算要注意:**每頁大約消耗 1,500–3,000 token(依內容密度而定),一份 50 頁的合約可能就要 75,000–150,000 token——這很快就會逼近 max_tokens 上限,或讓帳單超出預期。

長文件策略建議:

• 20 頁以內:直接整份丟進去,讓 Claude 自己讀
• 20–100 頁:先跑一次「目錄摘要」(只傳前幾頁或標題結構),再針對需要的章節拆段傳
• 100 頁以上:考慮用 Files API 上傳後取得 file_id,用 file_id 傳比每次重新 base64 更有效率;或結合第 7 課的 Prompt Caching(同一份文件重複問不同問題時,快取文件 token 可省 90% 費用)

PDF 上限:單一請求 32 MB;頁數上限依模型 context 而定——200K context 模型約 100 頁,1M context 的大模型最多 600 頁。超過就先拆分或改用 Files API。

Step 5:實戰小工具——收據辨識 + 串流輸出

把前面兩個技能組合起來:讀本機圖片(收據照片),傳給 Claude,用串流即時輸出辨識結果。

建立 receipt_ocr.py:

"""
收據辨識工具:接收圖片路徑,串流輸出結構化資訊
用法: python receipt_ocr.py <圖片路徑>
      python receipt_ocr.py receipt.jpg
"""

import sys
import base64
import anthropic
from dotenv import load_dotenv

load_dotenv()
client = anthropic.Anthropic()

# 副檔名 → MIME type 對照
MEDIA_TYPES = {
    "jpg":  "image/jpeg",
    "jpeg": "image/jpeg",
    "png":  "image/png",
    "webp": "image/webp",
    "gif":  "image/gif",
}

def analyze_receipt(image_path: str) -> None:
    # 判斷 media type
    ext = image_path.rsplit(".", 1)[-1].lower()
    media_type = MEDIA_TYPES.get(ext)
    if not media_type:
        print(f"不支援的格式:{ext}。請使用 jpg/png/webp/gif。")
        sys.exit(1)

    # 讀圖片並 base64 編碼
    with open(image_path, "rb") as f:
        image_data = base64.standard_b64encode(f.read()).decode("utf-8")

    print(f"正在分析 {image_path}...\n")
    print("=" * 40)

    # 串流呼叫
    with client.messages.stream(
        model="claude-haiku-4-5",      # 收據辨識 Haiku 夠用,最省錢
        max_tokens=512,
        system=(
            "你是一位收據分析助理。用繁體中文條列式回答,"
            "格式整齊、數字精確。若圖片模糊或資訊不清楚,如實說明。"
        ),
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": media_type,
                        "data": image_data,
                    }
                },
                {
                    "type": "text",
                    "text": (
                        "請從這張收據中提取以下資訊:\n"
                        "1. 店名 / 商家名稱\n"
                        "2. 日期與時間\n"
                        "3. 品項與各自金額\n"
                        "4. 小計 / 稅額 / 總金額\n"
                        "5. 付款方式(如有顯示)\n\n"
                        "若某項資訊在圖片中找不到,寫「未顯示」即可。"
                    )
                }
            ]
        }]
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)

    print("\n" + "=" * 40)

    # 顯示用量
    final = stream.get_final_message()
    print(f"token 消耗:輸入 {final.usage.input_tokens} / 輸出 {final.usage.output_tokens}")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: python receipt_ocr.py <圖片路徑>")
        sys.exit(1)
    analyze_receipt(sys.argv[1])

執行範例:

python receipt_ocr.py receipt.jpg

預期輸出(以實際收據為準):

正在分析 receipt.jpg...

========================================
1. 店名:全家便利商店 FamilyMart 信義店
2. 日期與時間:2026/07/04  14:32
3. 品項與金額:
   - 雞蛋沙拉三明治      $45
   - 茶裏王無糖綠茶 600ml $25
   - 關東煮(米血)        $15
4. 小計:$85 / 稅額:內含 / 總金額:$85
5. 付款方式:悠遊卡
========================================
token 消耗:輸入 1823 / 輸出 147

這個小工具已經可以直接整合進費用報帳系統、記帳 App 或內部審核流程——只要把 print 換成寫進資料庫,或改成串流回傳 HTTP Response,就是一個可以上線的 API。