Skip to Content
Wiki常见问题流式输出

流式输出 - 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