整个破解过程采用了自举模式，即先完成部分，然后再编译剩下的部分。我写了基本请求，然后让deepseek-R1自己看代码自己写，最后得到了可用的程序，期间删改了好多次，但是至少可用了现在。
使用方法，新建一个py文件然后粘贴进去运行即可，不会用python3的去问AI怎么用。两个studio_token都可以正常使用，我不知道缺了会怎么样，或者多久过期，先玩着吧。
import json
import random
import string
import requests
def generate_session_hash(length=12):
"""生成随机会话标识"""
return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))
def stream_ai_response(model_id: str, user_input: str, api_token: str, session_hash: str):
"""
流式获取AI模型响应的核心函数
参数：
- model_id: 模型标识（如 "deepseek-ai/DeepSeek-R1"）
- user_input: 用户输入文本
- api_token: API访问令牌
- session_hash: 会话标识
返回：
- 生成器，持续输出响应内容
"""
base_url = "https://ai-modelscope-deepseek-playground.ms.show"
# 初始化请求
init_url = f"{base_url}/gradio_api/queue/join"
headers = {
"Content-Type": "application/json",
"X-Studio-Token": api_token,
"Sec-Fetch-Site": "same-origin",
"Accept": "*/*"
}
# 构建请求体
payload = {
"data": [model_id, user_input, None],
"fn_index": 0,
"trigger_id": random.randint(1, 100),
"session_hash": session_hash,
"dataType": ["antdselect", "antdxsender", "state"]
}
try:
init_response = requests.post(
init_url,
headers=headers,
data=json.dumps(payload),
timeout=30
)
event_id = init_response.json().get("event_id")
except Exception as e:
raise ConnectionError(f"初始化请求失败: {str(e)}")
print("初始化成功，event_id:", event_id)
# 轮询获取流式响应
stream_url = f"{base_url}/gradio_api/queue/data"
params = {
"session_hash": session_hash,
"studio_token": api_token
}
try:
with requests.get(stream_url, headers=headers, params=params, stream=True) as response:
# current_thought = [] # 存储思考过程
current_answer = [] # 存储正式回答
for line in response.iter_lines():
if line:
decoded_line = line.decode('utf-8')
if decoded_line.startswith('data:'):
data = json.loads(decoded_line[5:].strip())
if data.get("msg") == "close_stream":
print("\n[聊天结束]")
yield ("[END]",current_answer) # 可选：发送结束标记
return # 关键修复：必须用return终止生成器
elif data.get("msg") == "process_generating":
content_updates = data.get("output", {}).get("data", [])
if len(content_updates) > 4:
content_block = content_updates[4]
for op in content_block:
if isinstance(op, list) and len(op) >= 3:
# 正式回答的增量内容
if op[0] == "append" and "content" in op[1][2]:
current_answer.append(op[2])
yield ("ANSWER", op[2])
# 思考过程元数据
elif op[0] == "add" and "reason_content" in op[1]:
# current_thought = [op[2]] # 重置思考内容
yield ("THINKING", op[2])
# 最终回答元数据
elif op[0] == "add" and "answer_content" in op[1]:
full_answer = op[2]
# 如果还有未输出的回答内容
if current_answer and not full_answer.startswith(''.join(current_answer)):
diff = full_answer[len(''.join(current_answer)):]
if diff:
current_answer.append(diff)
except Exception as e:
raise ConnectionError(f"流式响应中断: {str(e)}")
# 使用示例
if __name__ == "__main__":
print("启动")
session_hash = generate_session_hash(11)
print("生成的会话标识:", session_hash)
generator = stream_ai_response(
model_id="deepseek-ai/DeepSeek-R1",
user_input="hi",
api_token="4870862b-9a17-4e2b-950b-924b584f8117",
session_hash=session_hash
)
# studio_token: 0b45ecd9-c62b-46b2-b03f-914b1abc57e0
for content_type, content_chunk in generator:
if content_type == "THINKING":
print(f"\n[思考过程]\n{content_chunk}\n[正式回答]")
elif content_type == "ANSWER":
print(content_chunk, end='', flush=True)
else:
print("\n[其他内容]",content_chunk, end='', flush=True)
