流式输出

通过设置 stream: true 使用 Server-Sent Events（SSE）流式接收响应，降低首字延迟。

请求

curl https://api.aiqizhilian.tech/v1/messages \
  -H "Authorization: Bearer <你的 API Key>" \
  -H "Content-Type: application/json" \
  -H "anthropic-version: 2023-06-01" \
  -d '{
    "model": "claude-sonnet-4-6",
    "max_tokens": 256,
    "stream": true,
    "messages": [
      {
        "role": "user",
        "content": "用三句话介绍东京。"
      }
    ]
  }'

标准事件序列

一次完整的流式响应包含以下 SSE 事件，按顺序到达：

顺序	event	含义
1	message_start	消息开始，包含 message 元数据（id / model / role / usage 初值）
2	content_block_start	一个 content block 开始（type 可能是 text 或 tool_use）
3..N	content_block_delta	增量内容；text 块用 text_delta，tool_use 块用 input_json_delta
N+1	content_block_stop	当前 content block 结束
末	message_delta	携带最终 stop_reason、stop_sequence 和最终 usage
末	message_stop	流终结标志

如果响应包含多个 content block（例如先 text 后 tool_use），会循环 content_block_start → *_delta → content_block_stop。

示例

文本响应的 SSE 流（实测样例，最大 50 token）：

event: message_start
data: {"type":"message_start","message":{"id":"msg_xxx","model":"claude-sonnet-4-6","role":"assistant","content":[],"usage":{"input_tokens":12,"output_tokens":0}}}

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

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

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"是日本"}}

...

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":24}}

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

客户端解析要点

• 按 event: + data: 行成对读取。
• 读取 content_block_delta 内的 delta.text（text 块）或 delta.partial_json（tool_use 块）拼接。
• 收到 message_stop 时本次响应结束。
• 最终 usage（含完整 output_tokens 和 cache 相关字段）在 message_delta 中。

不要自己撸 SSE 解析，建议直接用：

• Python：anthropic 官方 SDK 或 httpx-sse
• Node.js：@anthropic-ai/sdk 或 eventsource-parser

SDK 示例

Python（anthropic 官方 SDK）

import anthropic

client = anthropic.Anthropic(
    base_url="https://api.aiqizhilian.tech",
    api_key="<你的 API Key>",
)

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=256,
    messages=[{"role": "user", "content": "用三句话介绍东京"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()
    print(f"\n[stop_reason] {final.stop_reason}  [usage] {final.usage}")

Node.js（@anthropic-ai/sdk）

import Anthropic from '@anthropic-ai/sdk'

const client = new Anthropic({
  baseURL: 'https://api.aiqizhilian.tech',
  apiKey: process.env.API_KEY,
})

const stream = await client.messages.stream({
  model: 'claude-sonnet-4-6',
  max_tokens: 256,
  messages: [{ role: 'user', content: '用三句话介绍东京' }],
})

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

const final = await stream.finalMessage()
console.log('\n[stop_reason]', final.stop_reason, '[usage]', final.usage)

注意事项

• 不支持 /v1/messages/count_tokens：当前网关未实现该端点（返回 404）。如果客户端有 token 预估需求，需自行用 tokenizer 估算或调用主请求获取 usage。
• 流式中断：HTTP/2 空闲超时或上游异常可能导致流提前结束。生产代码建议加超时保护和指数退避重试，但要避免对非幂等业务无脑重试。