# coding=utf-8 import requests import json import urllib3 from Config import * if __name__ == '__main__': # 禁止警告 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # 设置请求头和数据 headers = { 'Content-Type': 'application/json', 'Authorization': 'Bearer ' + HW_API_KEY # 把yourApiKey替换成真实的API Key } data = { "model": HW_MODEL_NAME, "messages": [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "什么水果不影响血糖值?"} ], "stream": True, # 启用流式输出 "temperature": 1.0 } # 初始化变量,用于保存完整的响应内容 full_response = "" # 发送请求并处理流式响应 with requests.post(HW_API_URL, headers=headers, data=json.dumps(data), verify=False, stream=True) as resp: if resp.status_code == 200: for line in resp.iter_lines(): if line: # 过滤掉空行 decoded_line = line.decode('utf-8') if decoded_line.startswith("data: "): # 处理以 "data: " 开头的行 json_str = decoded_line[len("data: "):] # 去掉 "data: " 前缀 if json_str == "[DONE]": # 流式响应结束 break try: # 解析 JSON 数据 json_data = json.loads(json_str) # 提取流式输出的内容 if "choices" in json_data and len(json_data["choices"]) > 0: # 检查是否有 "delta" 字段 delta = json_data["choices"][0].get("delta", {}) # 尝试从 "content" 或 "reasoning_content" 中提取内容 content = delta.get("content", delta.get("reasoning_content", "")) if content: # 实时逐字输出内容 print(content, end="", flush=True) # 将内容累积到完整响应中 full_response += content except json.JSONDecodeError: pass # 忽略解析错误 else: print("请求失败,状态码:", resp.status_code) print("响应内容:", resp.text)