96 lines
3.7 KiB
Python
96 lines
3.7 KiB
Python
import json
|
|
import requests
|
|
from Config.Config import GPTNB_API_KEY
|
|
|
|
class ModelInteractor:
|
|
def __init__(self, api_key=GPTNB_API_KEY, api_url="https://goapi.gptnb.ai/v1/chat/completions"):
|
|
self.api_key = api_key
|
|
self.api_url = api_url
|
|
self.headers = {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {self.api_key}"
|
|
}
|
|
|
|
def stream_request(self, model, prompt, temperature=0.7):
|
|
"""
|
|
发送流式请求到模型API
|
|
|
|
参数:
|
|
- model: 模型名称
|
|
- prompt: 用户提示词
|
|
- temperature: 温度参数，控制输出的随机性
|
|
|
|
返回:
|
|
- 无返回值，直接打印流式响应
|
|
"""
|
|
payload = {
|
|
"model": model,
|
|
"messages": [{
|
|
"role": "user",
|
|
"content": prompt
|
|
}],
|
|
"temperature": temperature,
|
|
"stream": True
|
|
}
|
|
|
|
try:
|
|
response = requests.post(
|
|
self.api_url,
|
|
headers=self.headers,
|
|
data=json.dumps(payload),
|
|
stream=True,
|
|
timeout=30
|
|
)
|
|
response.raise_for_status()
|
|
|
|
print(f"使用模型 {model} 的流式响应内容: ")
|
|
buffer = ""
|
|
for chunk in response.iter_content(chunk_size=None):
|
|
if chunk:
|
|
chunk_data = chunk.decode('utf-8', errors='replace')
|
|
buffer += chunk_data
|
|
|
|
# 处理buffer中的所有完整JSON对象
|
|
while True:
|
|
# 查找JSON对象的开始和结束位置
|
|
start_pos = buffer.find('{')
|
|
if start_pos == -1:
|
|
break # 没有找到JSON开始
|
|
|
|
# 尝试找到匹配的结束括号
|
|
depth = 1
|
|
end_pos = start_pos + 1
|
|
while end_pos < len(buffer) and depth > 0:
|
|
if buffer[end_pos] == '{':
|
|
depth += 1
|
|
elif buffer[end_pos] == '}':
|
|
depth -= 1
|
|
end_pos += 1
|
|
|
|
if depth == 0:
|
|
# 找到了完整的JSON对象
|
|
json_str = buffer[start_pos:end_pos]
|
|
buffer = buffer[end_pos:]
|
|
|
|
try:
|
|
data = json.loads(json_str)
|
|
if 'choices' in data and len(data['choices']) > 0:
|
|
delta = data['choices'][0].get('delta', {})
|
|
content = delta.get('content', '')
|
|
if content and content != '\n':
|
|
print(content, end='', flush=True)
|
|
except json.JSONDecodeError as e:
|
|
print(f"[调试] JSON解析错误: {e}, 内容: {json_str}")
|
|
else:
|
|
# 没有找到完整的JSON对象
|
|
break
|
|
|
|
# 检查是否结束
|
|
if 'data: [DONE]' in buffer:
|
|
print("\n流式响应结束")
|
|
return
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"请求发生错误: {e}")
|
|
|