huanghai
/
QingLong
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

'commit'

Browse Source
main
HuangHai 5 months ago
parent 46b8665528
commit 9d4f5b4f02
3 changed files with 114 additions and 105 deletions
Show Stats Download Patch File Download Diff File

122
AI/AiService/CallDeepSeekStream.py
Unescape View File

@ -2,22 +2,118 @@ import uvicorn
from fastapi import FastAPI, Body
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse, PlainTextResponse
import asyncio
import socket
from openai import OpenAI
from MarkdownToJsonUtil import *
import markdown_to_json
import json
import asyncio
# 阿里云中用来调用 deepseek v3 的密钥
MODEL_API_KEY = "sk-01d13a39e09844038322108ecdbd1bbc"
MODEL_NAME = "deepseek-v3"
#MODEL_NAME = "qwen-plus"
#MODEL_NAME = "deepseek-v3"
MODEL_NAME = "qwen-plus"
# 初始化 OpenAI 客户端
client = OpenAI(
api_key=MODEL_API_KEY,
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
def markdown_to_dict(markdown_content):
"""
将 Markdown 内容转换为 Python 字典
"""
# 将 Markdown 转换为 JSON 字符串
json_content = markdown_to_json.jsonify(markdown_content)
# 解码 Unicode 转义
json_content = json_content.encode('utf-8').decode('unicode_escape')
# 将 JSON 字符串转换为字典
return json.loads(json_content)
def extract_level1(json_dict):
"""
提取一级目录，生成指定格式的 JSON 对象列表
"""
# 获取第一个一级目录的名称
level1_title = next(iter(json_dict.keys()), None)
if level1_title:
return [{"type": "cover", "data": {"title": level1_title, "text": level1_title}}]
return []
def extract_level2_and_level3(json_dict, level1_title=None):
"""
提取指定一级目录下的二级目录及其三级目录内容，生成指定格式的 JSON 对象列表
"""
# 如果没有指定一级目录，则使用第一个一级目录
if level1_title is None:
level1_title = next(iter(json_dict.keys()), None)
if level1_title and level1_title in json_dict:
result = []
for level2_title, level2_content in json_dict[level1_title].items():
# 输出二级目录
result.append({"type": "transition", "data": {"title": level2_title, "text": level2_title}})
# 输出三级目录内容
if isinstance(level2_content, dict):
for level3_title, level3_items in level2_content.items():
# 确保 level3_items 是列表
if isinstance(level3_items, list):
items = [{"title": item, "text": item} for item in level3_items]
else:
items = [{"title": str(level3_items), "text": str(level3_items)}]
result.append({
"type": "content",
"data": {
"title": level3_title,
"items": items
}
})
return result
return []
def extract_contents(json_dict, level1_title=None):
"""
提取所有二级目录名称，生成目录部分的 JSON 对象
"""
# 如果没有指定一级目录，则使用第一个一级目录
if level1_title is None:
level1_title = next(iter(json_dict.keys()), None)
if level1_title and level1_title in json_dict:
# 获取所有二级目录名称
level2_titles = list(json_dict[level1_title].keys())
return {"type": "contents", "data": {"items": level2_titles}}
return {"type": "contents", "data": {"items": []}}
async def ConvertMarkdownToJson(markdown_content):
"""
生成一个 AsyncIterable，逐行返回 JSON 字符串
"""
# 将 Markdown 转换为字典
json_dict = markdown_to_dict(markdown_content)
# 提取一级目录
level1_json = extract_level1(json_dict)
for item in level1_json:
yield json.dumps(item, ensure_ascii=False)
await asyncio.sleep(0.5) # 控制逐行输出的速度
# 生成目录部分
contents_json = extract_contents(json_dict)
yield json.dumps(contents_json, ensure_ascii=False)
await asyncio.sleep(0.5)
# 提取二级目录及其三级目录内容
level2_and_level3_json = extract_level2_and_level3(json_dict)
for item in level2_and_level3_json:
yield json.dumps(item, ensure_ascii=False)
await asyncio.sleep(0.5)
# 添加结束标记
yield '{"type": "end" }'
# 获取本机所有 IPv4 地址
def get_local_ips():
ips = []
@ -35,25 +131,35 @@ def get_local_ips():
# 流式生成数据的函数
async def generate_stream_markdown(course_name: str):
"""
流式生成 Markdown 数据，并在控制台输出完整的 Markdown 内容
"""
# 调用阿里云 API，启用流式响应
stream = client.chat.completions.create(
model=MODEL_NAME,
messages=[
{'role': 'system', 'content': '你是一个教学经验丰富的基础教育教师'},
{'role': 'user', 'content': '帮我设计一下' + course_name + '的课件提纲，用markdown格式返回。不要返回 ```markdown 或者 ``` 这样的内容！'}
{'role': 'user', 'content': '帮我设计一下' + course_name + '的课件提纲，用markdown格式返回。强调1、标签只能返回 #,##,###,-,其它标签一率不可以返回，这个非常重要！2、不要返回 ```markdown 或者 ``` 这样的内容！ 3、每部分都有生成完整的一、二、三级内容，不能省略。4、一级、二级、三级都要同步生成一个对于当前部分的一句话描述。以【】包裹起来。'}
],
stream=True, # 启用流式响应
timeout=6000,
)
# 初始化完整的 Markdown 内容
full_markdown = ""
# 逐字返回数据
for chunk in stream:
if chunk.choices[0].delta.content:
for char in chunk.choices[0].delta.content:
chunk_content = chunk.choices[0].delta.content
full_markdown += chunk_content # 拼接 Markdown 内容
for char in chunk_content:
yield char.encode("utf-8")
await asyncio.sleep(0.05) # 控制逐字输出的速度
# 在控制台输出完整的 Markdown 内容
print("\n完整的 Markdown 内容：")
print(full_markdown)
app = FastAPI()
# 添加 CORS 中间件
@ -89,7 +195,7 @@ async def aippt_outline(
@app.post("/api/tools/aippt") # 修改为 POST 方法
async def aippt(content: str = Body(..., embed=True, description="Markdown 内容")): # 使用 Body 接收请求体参数
return StreamingResponse(
getMyJson(content), # 传入 content
ConvertMarkdownToJson(content), # 传入 content
media_type="text/plain", # 使用 text/plain 格式
headers={
"Cache-Control": "no-cache",

97
AI/AiService/MarkdownToJsonUtil.py
Unescape View File

@ -1,97 +0,0 @@
import markdown_to_json
import json
import asyncio
def markdown_to_dict(markdown_content):
"""
将 Markdown 内容转换为 Python 字典
"""
# 将 Markdown 转换为 JSON 字符串
json_content = markdown_to_json.jsonify(markdown_content)
# 解码 Unicode 转义
json_content = json_content.encode('utf-8').decode('unicode_escape')
# 将 JSON 字符串转换为字典
return json.loads(json_content)
def extract_level1(json_dict):
"""
提取一级目录，生成指定格式的 JSON 对象列表
"""
# 获取第一个一级目录的名称
level1_title = next(iter(json_dict.keys()), None)
if level1_title:
return [{"type": "cover", "data": {"title": level1_title, "text": level1_title}}]
return []
def extract_level2_and_level3(json_dict, level1_title=None):
"""
提取指定一级目录下的二级目录及其三级目录内容，生成指定格式的 JSON 对象列表
"""
# 如果没有指定一级目录，则使用第一个一级目录
if level1_title is None:
level1_title = next(iter(json_dict.keys()), None)
if level1_title and level1_title in json_dict:
result = []
for level2_title, level2_content in json_dict[level1_title].items():
# 输出二级目录
result.append({"type": "transition", "data": {"title": level2_title, "text": level2_title}})
# 输出三级目录内容
if isinstance(level2_content, dict):
for level3_title, level3_items in level2_content.items():
items = [{"title": item, "text": item} for item in level3_items]
result.append({
"type": "content",
"data": {
"title": level3_title,
"items": items
}
})
return result
return []
def extract_contents(json_dict, level1_title=None):
"""
提取所有二级目录名称，生成目录部分的 JSON 对象
"""
# 如果没有指定一级目录，则使用第一个一级目录
if level1_title is None:
level1_title = next(iter(json_dict.keys()), None)
if level1_title and level1_title in json_dict:
# 获取所有二级目录名称
level2_titles = list(json_dict[level1_title].keys())
return {"type": "contents", "data": {"items": level2_titles}}
return {"type": "contents", "data": {"items": []}}
async def getMyJson(markdown_content):
"""
生成一个 AsyncIterable，逐行返回 JSON 字符串
"""
# 将 Markdown 转换为字典
json_dict = markdown_to_dict(markdown_content)
# 提取一级目录
level1_json = extract_level1(json_dict)
for item in level1_json:
yield json.dumps(item, ensure_ascii=False)
await asyncio.sleep(0.5) # 控制逐行输出的速度
# 生成目录部分
contents_json = extract_contents(json_dict)
yield json.dumps(contents_json, ensure_ascii=False)
await asyncio.sleep(0.5)
# 提取二级目录及其三级目录内容
level2_and_level3_json = extract_level2_and_level3(json_dict)
for item in level2_and_level3_json:
yield json.dumps(item, ensure_ascii=False)
await asyncio.sleep(0.5)
# 添加结束标记
yield '{"type": "end" }'

BIN
AI/AiService/__pycache__/MarkdownToJsonUtil.cpython-310.pyc
View File

Binary file not shown.
Write Preview
Loading…
Cancel
Save