/串流回應

串流回應

串流回應（Streaming）允許你在模型生成過程中即時接收輸出，提升使用者體驗和感知速度。

工作原理#

诺玛AI 使用 Server-Sent Events (SSE) 協議實作串流回應：

客戶端傳送請求時設定 `stream: true`
伺服器逐步回傳生成的內容片段（chunk）
每個 chunk 以 `data: ` 字首透過 SSE 傳送
生成結束時傳送 `data: [DONE]`

OpenAI 協議串流#

cURL

程式碼

curl https://as.apinoma.com/v1/chat/completions \
  -H "Authorization: Bearer $APINOMA_API_KEY" \
  -H "Content-Type: application/json" \
  -N \
  -d '{
    "model": "openai/gpt-4o",
    "messages": [{"role": "user", "content": "寫一首關於程式設計的詩"}],
    "stream": true
  }'

Python

TypeScript

Anthropic 協議串流#

Python

程式碼

import anthropic
 
client = anthropic.Anthropic(
    base_url="https://as.apinoma.com/anthropic",
    api_key="<你的 APINOMA_API_KEY>"
)
 
with client.messages.stream(
    model="anthropic/claude-sonnet-4.6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "寫一首關於程式設計的詩"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

TypeScript

串流 + Function Calling#

串流回應也支援函式呼叫場景。模型會先串流輸出工具呼叫請求，你處理完成後繼續對話：

程式碼

stream = client.chat.completions.create(
    model="openai/gpt-4o",
    messages=[{"role": "user", "content": "今天北京天氣怎麼樣？"}],
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "獲取指定城市的天氣",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名稱"}
                },
                "required": ["city"]
            }
        }
    }],
    stream=True
)
 
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.tool_calls:
        # 處理工具呼叫
        print(f"呼叫工具: {delta.tool_calls[0].function}")
    elif delta.content:
        print(delta.content, end="", flush=True)

錯誤處理和重連#

串流連線可能因網路問題中斷。建議實作重連邏輯。

程式碼

import time
 
def stream_with_retry(client, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            stream = client.chat.completions.create(stream=True, **kwargs)
            for chunk in stream:
                yield chunk
            return  # 成功完成
        except Exception as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # 指數退避
                print(f"\n連線中斷，{wait}s 後重試...")
                time.sleep(wait)
            else:
                raise e

最佳實踐#

始終設定超時 — 避免無限等待
處理不完整的 chunk — 某些 chunk 可能沒有 content
實作重連機制 — 使用指數退避策略
前端使用 `flush` — 確保內容即時顯示

最後更新於 2026 年 6 月 23 日