流式输出 - MyTokenGate
1. 在 python 中使用流式输出
1.1 基于 OpenAI 库的流式输出
在一般场景中,推荐您使用 OpenAI 的库进行流式输出。
from openai import OpenAI
client = OpenAI(
base_url='https://gateway.mytokengate.com/v1',
api_key='your-api-key'
)
# 发送带有流式输出的请求
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": "请介绍一下人工智能的发展历史"}
],
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)1.2 基于 requests 库的流式输出
如果您有非 openai 的场景,如您需要基于 request 库使用 API,请注意:除了 payload 中的 stream 需要设置外,request 请求的参数也需要设置 stream = True,才能正常按照 stream 模式进行返回。
import requests
import json
url = "https://gateway.mytokengate.com/v1/chat/completions"
payload = {
"model": "gpt-4o",
"messages": [
{
"role": "user",
"content": "请介绍一下人工智能的发展历史"
}
],
"stream": True # 此处需要设置为 stream 模式
}
headers = {
"accept": "application/json",
"content-type": "application/json",
"authorization": "Bearer your-api-key"
}
response = requests.post(url, json=payload, headers=headers, stream=True) # 此处 request 需要指定 stream 模式
# 打印流式返回信息
if response.status_code == 200:
full_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', '')
if content:
print(content, end="", flush=True)
full_content += content
else:
print(f"请求失败,状态码:{response.status_code}")2. curl 中使用流式输出
curl 命令的处理机制默认情况下,curl 会缓冲输出流,所以即使服务器分块(chunk)发送数据,也需要等缓冲区填满或连接关闭后才看到内容。传入 -N(或 —no-buffer)选项,可以禁止此缓冲,让数据块立即打印到终端,从而实现流式输出。
curl -N -s \
--request POST \
--url https://gateway.mytokengate.com/v1/chat/completions \
--header 'Authorization: Bearer your-api-key' \
--header 'Content-Type: application/json' \
--data '{
"model": "gpt-4o",
"messages": [
{"role":"user","content":"有诺贝尔数学奖吗?"}
],
"stream": true
}'Last updated on