在一般场景中,推荐您使用 openai 的库进行流式输出。
from openai import OpenAI client = OpenAI( base_url='https://api.ap.siliconflow.com/v1', api_key='your-api-key' ) # 发送带有流式输出的请求 response = client.chat.completions.create( model="deepseek-ai/DeepSeek-V2.5", messages=[ {"role": "user", "content": "DeepSeek 发布,对大模型领域意味着什么?"} ], stream=True # 启用流式输出 ) # 逐步接收并处理响应 for chunk in response: chunk_message = chunk.choices[0].delta.content print(chunk_message, end='', flush=True)
如果您有非 openai 的场景,如您需要基于 request 库使用 siliconcloud API,请您注意: 除了 payload 中的 stream 需要设置外,request 请求的参数也需要设置 stream = True, 才能正常按照 stream 模式进行返回。
import requests url = "https://api.ap.siliconflow.com/v1/chat/completions" payload = { "model": "deepseek-ai/DeepSeek-V2.5", # 替换成你的模型 "messages": [ { "role": "user", "content": "DeepSeek 发布,对大模型领域意味着什么?" } ], "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: for chunk in response.iter_content(chunk_size=8192): if chunk: decoded_chunk = chunk.decode('utf-8') print(decoded_chunk, end='') else: print('Request failed with status code:', response.status_code)