流式传输
两种格式的服务器发送事件——事件结构不同,混用会导致看似没有任何响应。
设置 stream 设为 true,响应就会变为 text/event-stream。两个端点都支持流式响应,但它们输出的 事件结构不同.
Anthropic 格式
curl -N https://aiprimetech.io/v1/messages \
-H "x-api-key: $CLAUDEAPIKEY" \
-H "anthropic-version: 2023-06-01" \
-H "content-type: application/json" \
-d '{"model":"claude-sonnet-4-6","max_tokens":512,"stream":true,
"messages":[{"role":"user","content":"Count to five"}]}'
event: message_start
data: {"type":"message_start","message":{"id":"msg_...","usage":{"input_tokens":9}}}
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"One"}}
event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":14}}
event: message_stop
data: {"type":"message_stop"}
文本通过以下内容传入: content_block_delta 事件,位置为 delta.text。最终输出 token 数通过以下内容传入: message_delta,而不是 message_start.
OpenAI 格式
data: {"choices":[{"delta":{"content":"One"},"index":0}]}
data: {"choices":[{"delta":{},"finish_reason":"stop","index":0}]}
data: [DONE]
文本传入的位置为 choices[0].delta.content,流以字面值 data: [DONE] 作为结束标记,该标记不是 JSON。
这种不匹配会导致响应为空且不报错:客户端解析
choices[].delta.content 时,如果接收的是 Messages 流,每次都找不到任何内容,却仍报告成功。如果流式响应看似正常,但没有返回任何内容,请检查你正在解析哪种结构。Python,Anthropic SDK
from anthropic import Anthropic
client = Anthropic(api_key=class="s">"sk-your-key", base_url=class="s">"https://aiprimetech.io")
with client.messages.stream(
model=class="s">"claude-sonnet-4-6",
max_tokens=512,
messages=[{class="s">"role": class="s">"user", class="s">"content": class="s">"Count to five"}],
) as stream:
for text in stream.text_stream:
print(text, end=class="s">"", flush=True)
操作说明
- 使用
curl -N用于手动测试,否则 curl 会缓冲整个流,看起来就像没有任何响应。 - 禁用应用前端代理的缓冲(
proxy_buffering off在 nginx 中),否则客户端会在结束时一次性收到全部回复。 - 流式传输中途断开连接,仍会对已经生成的 token 收费。
- 流式响应不会改变价格,只会改变字节到达的时间。