Source: https://aiprimetech.io/zh/docs/api-reference/streaming/

[首页](/zh/) / [文档](/zh/docs/) / 流式传输 

# 流式传输 

两种格式的服务器发送事件——事件结构不同，混用会导致看似没有任何响应。 

 复制页面 [ 以 Markdown 查看](/zh/docs/api-reference/streaming.md) [ 向 ChatGPT 提问](https://chatgpt.com/?q=Read%20https%3A//aiprimetech.io/docs/api-reference/streaming.md%20and%20answer%20questions%20about%20it.%20Start%20with%20a%20short%20summary.) [ 在 Claude 中提问](https://claude.ai/new?q=Read%20https%3A//aiprimetech.io/docs/api-reference/streaming.md%20and%20answer%20questions%20about%20it.%20Start%20with%20a%20short%20summary.)   

设置 stream 设为 true，响应就会变为 text/event-stream。两个端点都支持流式响应，但它们输出的 事件结构不同. 

## Anthropic 格式 

```
Copycurl -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"}]}'
```

```
Copyevent: 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 格式 

```
Copydata: {"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 

```
Copyfrom 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 收费。
- 流式响应不会改变价格，只会改变字节到达的时间。 

[消息 API非流式响应参考](/zh/docs/api-reference/messages/)[错误流式传输中途失败](/zh/docs/api-reference/errors/)[故障排查空响应](/zh/docs/guides/troubleshooting/) 

[上一页模型端点](/zh/docs/api-reference/models-endpoint/)[下一页错误](/zh/docs/api-reference/errors/)