|
|
# -*- coding: utf-8 -*-
|
|
|
import time
|
|
|
from pathlib import Path
|
|
|
from typing import Iterator
|
|
|
import dashscope
|
|
|
from dashscope import Generation
|
|
|
from dashscope.api_entities.dashscope_response import DashScopeAPIResponse
|
|
|
|
|
|
|
|
|
class TestDeepSeek:
|
|
|
# 模型版本常量
|
|
|
R1 = "deepseek-r1"
|
|
|
V3 = "deepseek-v3"
|
|
|
# API密钥（建议从环境变量读取）
|
|
|
API_KEY = "sk-01d13a39e09844038322108ecdbd1bbc"
|
|
|
|
|
|
@staticmethod
|
|
|
def get_system_prompt() -> str:
|
|
|
"""读取系统提示文件"""
|
|
|
md_path = Path(r"D:\dsWork\QingLong\PptGenerator\md-file\readme\default.md")
|
|
|
if not md_path.exists():
|
|
|
raise FileNotFoundError(f"提示文件不存在: {md_path}")
|
|
|
return md_path.read_text(encoding='utf-8')
|
|
|
|
|
|
@classmethod
|
|
|
def call_with_stream(cls) -> Iterator[DashScopeAPIResponse]:
|
|
|
"""流式调用API"""
|
|
|
system_prompt = (
|
|
|
"返回的教案内容请严格按照指定的markdown语法格式返回，用来生成ppt。"
|
|
|
"需要注意:1、最开头的Title:..,Author: ... Date:... 不能省略掉，有用处。"
|
|
|
"2、第一级用#,第二级用##,第二级是独立的PPT一个页，每页中的数据必须有> 开头,"
|
|
|
"如果是页面中的条目，必须用 - 开头,结果不要返回图片！请仔细阅读参考格式："
|
|
|
) + cls.get_system_prompt()
|
|
|
|
|
|
try:
|
|
|
return Generation.call(
|
|
|
model=cls.R1,
|
|
|
api_key=cls.API_KEY,
|
|
|
messages=[
|
|
|
{"role": "system", "content": system_prompt},
|
|
|
{"role": "user", "content": "请帮我生成一个小学数学内角和的教案，按标准格式输出。"}
|
|
|
],
|
|
|
result_format='message',
|
|
|
stream=True, # 启用流式模式
|
|
|
incremental_output=True # 启用增量输出
|
|
|
)
|
|
|
except Exception as e:
|
|
|
raise RuntimeError(f"API调用失败: {str(e)}") from e
|
|
|
|
|
|
@classmethod
|
|
|
def main(cls):
|
|
|
"""主执行方法（带实时进度提示）"""
|
|
|
print("🔄 开始生成教案，请稍候...")
|
|
|
start_time = time.time()
|
|
|
last_update = start_time
|
|
|
spinner = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏']
|
|
|
spinner_idx = 0
|
|
|
|
|
|
try:
|
|
|
responses = cls.call_with_stream()
|
|
|
full_content = []
|
|
|
has_content = False
|
|
|
|
|
|
for response in responses:
|
|
|
current_time = time.time()
|
|
|
# 更新进度动画（每秒更新）
|
|
|
if current_time - last_update > 0.1:
|
|
|
print(f"\r{spinner[spinner_idx % 10]} 生成中 [{int(current_time - start_time)}s]", end="")
|
|
|
spinner_idx += 1
|
|
|
last_update = current_time
|
|
|
|
|
|
# 处理响应内容
|
|
|
if response.status_code == 200 and response.output:
|
|
|
chunk = response.output.choices[0].message.content
|
|
|
if chunk:
|
|
|
full_content.append(chunk)
|
|
|
if not has_content:
|
|
|
print("\n\n✅ 开始接收内容：")
|
|
|
has_content = True
|
|
|
print(chunk, end="", flush=True) # 实时显示内容
|
|
|
|
|
|
# 最终处理
|
|
|
if full_content:
|
|
|
md_content = ''.join(full_content)
|
|
|
print(f"\n\n🎉 生成完成！总耗时: {int(time.time() - start_time)}秒")
|
|
|
|
|
|
# 保存文件
|
|
|
output_path = Path(r"D:\dsWork\QingLong\PptGenerator\md-file\readme\5.md")
|
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
output_path.write_text(md_content, encoding='utf-8')
|
|
|
print(f"💾 结果已保存至: {output_path}")
|
|
|
else:
|
|
|
print("\n⚠️ 未收到有效内容，请检查提示词和API配置")
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"\n\n❌ 发生错误: {str(e)}")
|
|
|
print("请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code")
|
|
|
finally:
|
|
|
print("\n🔚 程序执行结束")
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
dashscope.api_key = TestDeepSeek.API_KEY
|
|
|
TestDeepSeek.main() |
