python学习 调用硅基流动 对话模型接口 实现 流式输出

效果：

代码：

import requests
import jsonurl = "https://api.siliconflow.cn/v1/chat/completions"payload = {"model": "Qwen/Qwen3-8B","max_tokens": 512,"enable_thinking": False,"thinking_budget": 4096,"min_p": 0.05,"temperature": 0.7,"top_p": 0.7,"top_k": 50,"frequency_penalty": 0.5,"n": 1,"stream": True,  # 启用流式输出"messages": [{"content": "武汉得气候怎么样","role": "system"}]
}
#xxxxxxx换成自己得密钥
headers = {"Authorization": "Bearer xxxxxxx","Content-Type": "application/json"
}# 发送流式请求
response = requests.post(url, json=payload, headers=headers, stream=True)# 处理流式响应
if response.status_code == 200:print("开始接收流式输出:")print("-" * 40)for line in response.iter_lines():if line:# 解码并去除 "data: " 前缀decoded_line = line.decode('utf-8')if decoded_line.startswith('data: '):data_str = decoded_line[6:]  # 去除 "data: " 前缀# 检查是否为结束标志if data_str.strip() == '[DONE]':print("\n\n流式输出完成")breaktry:# 解析 JSON 数据data = json.loads(data_str)# 提取内容并实时输出if 'choices' in data and len(data['choices']) > 0:delta = data['choices'][0].get('delta', {})content = delta.get('content', '')if content:print(content, end='', flush=True)except json.JSONDecodeError:# 跳过无法解析的行continue
else:print(f"请求失败，状态码: {response.status_code}")print(response.text)

解释

# 普通请求：一次性返回完整响应
response = requests.post(url, json=payload)  # 等待完整响应

# 流式请求：保持连接开放，持续接收数据
response = requests.post(url, json=payload, stream=True)  # 立即返回，持续接收

sse格式：

data: {"choices":[{"delta":{"content":"武"}}]}

data: {"choices":[{"delta":{"content":"汉"}}]}

data: {"choices":[{"delta":{"content":"的"}}]}

data: [DONE]