流式输出

流式输出 (Streaming) 让模型响应实时返回,逐块 (chunk) 输出,适用于聊天、长文本生成等需要即时反馈的场景。

1. 在 Python 中使用流式输出

1.1 基于 OpenAI 库的流式输出

在一般场景中,推荐您使用 OpenAI 的库进行流式输出:

from openai import OpenAI

client = OpenAI(
    base_url='https://khb.net.cn/v1',
    api_key='your-api-key'
)

response = client.chat.completions.create(
    model="Pro/zai-org/GLM-5.1",
    messages=[
        {"role": "user", "content": "KHB 算力平台上线,助力开发者轻松接入大模型能力。这对于整个大模型应用领域带来哪些改变?"}
    ],
    stream=True
)

for chunk in response:
    if not chunk.choices:
        continue
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
    if chunk.choices[0].delta.reasoning_content:
        print(chunk.choices[0].delta.reasoning_content, end="", flush=True)

1.2 基于 requests 库的流式输出

如果您有非 OpenAI 的场景,需要基于 requests 库调用 KHB API,请注意:除了 payload 中的 stream 需要设置外,request 请求的参数也需要设置 stream = True,才能正常按 stream 模式返回。

import requests
import json

url = "https://khb.net.cn/v1/chat/completions"

payload = {
    "model": "Pro/zai-org/GLM-5.1",
    "messages": [
        {
            "role": "user",
            "content": "KHB 算力平台上线,助力开发者轻松接入大模型能力。这对于整个大模型应用领域带来哪些改变?"
        }
    ],
    "stream": True
}

headers = {
    "accept": "application/json",
    "content-type": "application/json",
    "authorization": "Bearer your-api-key"
}

response = requests.post(url, json=payload, headers=headers, stream=True)

if response.status_code == 200:
    full_content = ""
    full_reasoning_content = ""
    for chunk in response.iter_lines():
        if chunk:
            chunk_str = chunk.decode('utf-8').replace('data: ', '')
            if chunk_str != "[DONE]":
                chunk_data = json.loads(chunk_str)
                delta = chunk_data['choices'][0].get('delta', {})
                content = delta.get('content', '')
                reasoning_content = delta.get('reasoning_content', '')
                if content:
                    print(content, end="", flush=True)
                    full_content += content
                if reasoning_content:
                    print(reasoning_content, end="", flush=True)
                    full_reasoning_content += reasoning_content
else:
    print(f"请求失败,状态码: {response.status_code}")

2. curl 中使用流式输出

curl 命令的处理机制默认情况下会缓冲输出流,所以即使服务器分块 (chunk) 发送数据,也需要等缓冲区填满或连接关闭后才看到内容。传入 -N (或 --no-buffer) 选项,可以禁止此缓冲,让数据块立即打印到终端,从而实现流式输出。

curl -N -s   --request POST   --url https://khb.net.cn/v1/chat/completions   --header 'Authorization: Bearer token'   --header 'Content-Type: application/json'   --data '{
    "model": "Qwen/Qwen2.5-72B-Instruct",
    "messages": [
      {"role":"user","content":"有诺贝尔数学奖吗?"}
    ],
    "stream": true
  }'