Streaming Claude API — Nhận Phản Hồi Thời Gian Thực Từ AI

Khi tạo một Message với Claude API, bạn có thể bật "stream": true để nhận response theo dạng server-sent events (SSE) — tức là nội dung được trả về từng phần một theo thời gian thực, thay vì phải chờ toàn bộ response hoàn chỉnh.

Với doanh nghiệp SME Việt Nam, đây là tính năng quan trọng để xây dựng trải nghiệm người dùng tốt hơn: thay vì màn hình trống trắng chờ 10-30 giây, người dùng thấy câu trả lời hiện ra từng chữ một — giống như đang chat trực tiếp với AI.

Streaming với SDK

Python SDK và TypeScript SDK cung cấp nhiều cách streaming. PHP SDK hỗ trợ streaming qua createStream(). Python SDK cho phép cả sync và async streams.

Ví dụ cơ bản — Python

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Xin chào!"}],
    model="claude-opus-4-6",
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Ví dụ cơ bản — TypeScript

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

const client = new Anthropic();

await client.messages
  .stream({
    messages: [{ role: "user", content: "Xin chào!" }],
    model: "claude-opus-4-6",
    max_tokens: 1024
  })
  .on("text", (text) => {
    process.stdout.write(text);
  });

Ví dụ — PHP

<?php

use Anthropic\Client;

$client = new Client(apiKey: getenv("ANTHROPIC_API_KEY"));

$stream = $client->messages->createStream(
    maxTokens: 1024,
    messages: [
        ['role' => 'user', 'content' => 'Xin chào!']
    ],
    model: 'claude-opus-4-6',
);

foreach ($stream as $message) {
    echo $message;
}

Ví dụ — Ruby

require "anthropic"

client = Anthropic::Client.new

stream = client.messages.stream(
  model: "claude-opus-4-6",
  max_tokens: 1024,
  messages: [{ role: "user", content: "Xin chào!" }]
)

stream.text.each { |text| print(text) }

Nhận Message Hoàn Chỉnh Không Cần Xử Lý Từng Event

Nếu không cần xử lý text ngay khi đến, SDK cung cấp cách dùng streaming ngầm nhưng vẫn trả về object Message hoàn chỉnh — giống như .create() trả về. Điều này đặc biệt hữu ích với các request có max_tokens lớn, nơi SDK yêu cầu streaming để tránh HTTP timeout.

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    max_tokens=128000,
    messages=[{"role": "user", "content": "Viết một phân tích chi tiết..."}],
    model="claude-opus-4-6",
) as stream:
    message = stream.get_final_message()

print(message.content[0].text)

Trong TypeScript, dùng finalMessage():

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

const client = new Anthropic();

const stream = client.messages.stream({
  max_tokens: 128000,
  messages: [{ role: "user", content: "Viết một phân tích chi tiết..." }],
  model: "claude-opus-4-6"
});

const message = await stream.finalMessage();
const textBlock = message.content.find((block) => block.type === "text");
if (textBlock && textBlock.type === "text") {
  console.log(textBlock.text);
}

Giải thích: .stream() giữ kết nối HTTP sống với server-sent events, sau đó .get_final_message() (Python) hoặc .finalMessage() (TypeScript) tích lũy tất cả event và trả về object Message hoàn chỉnh.

Các Loại Event (Event Types)

Mỗi server-sent event bao gồm tên event type và dữ liệu JSON kèm theo. Luồng stream sử dụng thứ tự event sau:

1. message_start: Chứa object Message với content rỗng.
2. Một loạt content blocks, mỗi block có:
   • content_block_start
   • Một hoặc nhiều event content_block_delta
   • content_block_stop
   Mỗi content block có index tương ứng với vị trí trong mảng content của Message cuối.
3. Một hoặc nhiều event message_delta, báo hiệu thay đổi cấp cao trong object Message cuối.
4. Event message_stop cuối cùng.

Ping Events

Event streams cũng có thể bao gồm bất kỳ số lượng event ping nào — đây là event keepalive để giữ kết nối sống.

Error Events

API đôi khi gửi lỗi trong event stream. Ví dụ, trong giai đoạn tải cao, bạn có thể nhận được overloaded_error:

event: error
data: {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}

Event Types khác

Theo chính sách versioning, có thể thêm event types mới. Code của bạn nên xử lý gracefully các event type không biết (không crash khi gặp event lạ).

Content Block Delta Types

Mỗi event content_block_delta chứa delta thuộc một type cụ thể, cập nhật content block tại index cho trước.

Text Delta

event: content_block_delta
data: {"type": "content_block_delta","index": 0,"delta": {"type": "text_delta", "text": "ello frien"}}

Input JSON Delta (dành cho Tool Use)

Delta của các content block tool_use tương ứng với cập nhật cho trường input của block. Để hỗ trợ granularity tối đa, các delta là chuỗi JSON cục bộ, trong khi tool_use.input cuối cùng luôn là object.

event: content_block_delta
data: {"type": "content_block_delta","index": 1,"delta": {"type": "input_json_delta","partial_json": "{\"location\": \"Ho Chi Minh Ci"}}

Thinking Delta (Extended Thinking)

Khi dùng Extended Thinking với streaming bật, bạn sẽ nhận thinking content qua thinking_delta events:

event: content_block_delta
data: {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "Tôi cần tìm GCD của 1071 và 462 dùng thuật toán Euclidean.\n\n1071 = 2 × 462 + 147"}}

Ngay trước content_block_stop, một event signature_delta đặc biệt được gửi để xác minh tính toàn vẹn của thinking block:

event: content_block_delta
data: {"type": "content_block_delta", "index": 0, "delta": {"type": "signature_delta", "signature": "EqQBCgIYAhIM1gbcDa9GJwZA2b3hGgxBdjrkzLoky3dl1pkiMOYds..."}}

HTTP Stream Response Đầy Đủ

Hãy dùng SDK khi streaming. Tuy nhiên nếu xây dựng tích hợp API trực tiếp, bạn cần xử lý các event này thủ công.

Cấu trúc stream response:

1. Event message_start
2. Nhiều content blocks, mỗi block chứa:
   • Event content_block_start
   • Nhiều event content_block_delta
   • Event content_block_stop
3. Event message_delta
4. Event message_stop

Ví dụ Request Streaming Cơ Bản

curl https://api.anthropic.com/v1/messages \
     --header "anthropic-version: 2023-06-01" \
     --header "content-type: application/json" \
     --header "x-api-key: $ANTHROPIC_API_KEY" \
     --data '{
  "model": "claude-opus-4-6",
  "messages": [{"role": "user", "content": "Xin chào!"}],
  "max_tokens": 256,
  "stream": true
}'

Ví dụ response stream:

event: message_start
data: {"type": "message_start", "message": {"id": "msg_1nZdL29xx5MUA1yADyHTEsnR8uuvGzszyY", "type": "message", "role": "assistant", "content": [], "model": "claude-opus-4-6", "stop_reason": null, "stop_sequence": null, "usage": {"input_tokens": 25, "output_tokens": 1}}}

event: content_block_start
data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}

event: ping
data: {"type": "ping"}

event: content_block_delta
data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Xin"}}

event: content_block_delta
data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": " chào!"}}

event: content_block_stop
data: {"type": "content_block_stop", "index": 0}

event: message_delta
data: {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence":null}, "usage": {"output_tokens": 15}}

event: message_stop
data: {"type": "message_stop"}

Streaming với Tool Use

Ví dụ request streaming với tool use — yêu cầu Claude dùng tool để báo thời tiết:

import anthropic

client = anthropic.Anthropic()

tools = [
    {
        "name": "get_weather",
        "description": "Lấy thông tin thời tiết hiện tại tại địa điểm cho trước",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "Thành phố và quốc gia, ví dụ: Hà Nội, Việt Nam",
                }
            },
            "required": ["location"],
        },
    }
]

with client.messages.stream(
    model="claude-opus-4-6",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "any"},
    messages=[
        {"role": "user", "content": "Thời tiết ở TP.HCM hôm nay thế nào?"}
    ],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Ví dụ TypeScript với streaming và tool use:

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

const client = new Anthropic();

const tools: Anthropic.Tool[] = [
  {
    name: "get_weather",
    description: "Lấy thông tin thời tiết hiện tại tại địa điểm cho trước",
    input_schema: {
      type: "object",
      properties: {
        location: {
          type: "string",
          description: "Thành phố, ví dụ: Hà Nội, TP.HCM"
        }
      },
      required: ["location"]
    }
  }
];

const stream = client.messages.stream({
  model: "claude-opus-4-6",
  max_tokens: 1024,
  tools: tools,
  tool_choice: { type: "any" },
  messages: [
    {
      role: "user",
      content: "Thời tiết ở Đà Nẵng hôm nay thế nào?"
    }
  ]
});

for await (const event of stream) {
  if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
    process.stdout.write(event.delta.text);
  }
}

Xử lý kết quả Tool Use trong Streaming

Khi stream với tool use, bạn cần tích lũy các input_json_delta events để xây dựng đầu vào tool hoàn chỉnh:

import json
import anthropic

client = anthropic.Anthropic()

def process_stream_with_tools(response_stream):
    tool_calls = []
    current_tool_input = {}
    current_tool_id = None

    for event in response_stream:
        if event.type == "content_block_start":
            if event.content_block.type == "tool_use":
                current_tool_id = event.content_block.id
                current_tool_input = {"partial_json": ""}

        elif event.type == "content_block_delta":
            if event.delta.type == "input_json_delta":
                current_tool_input["partial_json"] += event.delta.partial_json
            elif event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)

        elif event.type == "content_block_stop":
            if current_tool_id and current_tool_input.get("partial_json"):
                tool_calls.append({
                    "id": current_tool_id,
                    "input": json.loads(current_tool_input["partial_json"])
                })
                current_tool_id = None
                current_tool_input = {}

    return tool_calls

Best Practices cho Streaming

• Luôn xử lý unknown event types: API có thể thêm event types mới, code cần graceful handling.
• Dùng SDK khi có thể: SDK xử lý nhiều edge case phức tạp — error handling, reconnection, event accumulation.
• Token count trong message_delta là lũy tích: Đừng cộng dồn các số này, chỉ lấy giá trị cuối cùng.
• Ping events là bình thường: Đừng xử lý chúng như lỗi — chỉ là keepalive từ server.
• Với max_tokens lớn, bắt buộc dùng streaming: Tránh HTTP timeout khi tạo nội dung dài.