diff --git a/AI/AiService/CallDeepSeekStream.py b/AI/AiService/CallDeepSeekStream.py new file mode 100644 index 00000000..f8cd78b0 --- /dev/null +++ b/AI/AiService/CallDeepSeekStream.py @@ -0,0 +1,93 @@ +from fastapi import FastAPI +from fastapi.responses import StreamingResponse, PlainTextResponse +import socket +from openai import OpenAI + +app = FastAPI() + +# 阿里云中用来调用 deepseek v3 的密钥 +MODEL_API_KEY = "sk-01d13a39e09844038322108ecdbd1bbc" +MODEL_NAME = "deepseek-v3" + +# 初始化 OpenAI 客户端 +client = OpenAI( + api_key=MODEL_API_KEY, + base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", +) + +# 获取本机所有 IPv4 地址 +def get_local_ips(): + ips = [] + hostname = socket.gethostname() + try: + # 获取所有 IP 地址 + addrs = socket.getaddrinfo(hostname, None, family=socket.AF_INET) # 只获取 IPv4 地址 + for addr in addrs: + ip = addr[4][0] + if ip not in ips: + ips.append(ip) + except Exception as e: + print(f"获取 IP 地址失败: {e}") + return ips + +# 流式生成数据的函数 +async def generate_stream(course_name: str): + # 调用阿里云 API,启用流式响应 + stream = client.chat.completions.create( + model=MODEL_NAME, + messages=[ + {'role': 'system', 'content': '你是一个教学经验丰富的基础教育教师'}, + {'role': 'user', 'content': '帮我设计一下' + course_name + '的课件提纲,用markdown格式返回。'} + ], + stream=True, # 启用流式响应 + timeout=6000, + ) + + # 逐字返回数据 + for chunk in stream: + if chunk.choices[0].delta.content: + for char in chunk.choices[0].delta.content: + yield char.encode("utf-8") + await asyncio.sleep(0.05) # 控制逐字输出的速度 + +# 根路由,返回提示信息 +@app.get("/") +def root(): + return PlainTextResponse("Hello ApiStream") + +# 流式返回数据 +@app.get("/stream") +async def stream_data(course_name: str): + if not course_name: + return PlainTextResponse("请提供课程名称,例:course_name=三角形面积") + + # 返回流式响应 + return StreamingResponse( + generate_stream(course_name), + media_type="text/event-stream", # 使用 text/event-stream + headers={ + "Cache-Control": "no-cache", # 禁用缓存 + "Connection": "keep-alive", # 保持连接 + "Access-Control-Allow-Origin": "*", # 允许跨域 + "X-Accel-Buffering": "no" # 禁用 Nginx 缓冲(如果有代理) + } + ) + +# 运行应用 +if __name__ == "__main__": + import uvicorn + import asyncio + + # 获取本机所有 IPv4 地址 + ips = get_local_ips() + if not ips: + print("无法获取本机 IP 地址,使用默认地址 127.0.0.1") + ips = ["127.0.0.1"] + + # 打印所有 IP 地址 + print("服务将在以下 IP 地址上运行:") + for ip in ips: + print(f"http://{ip}:8000") + + # 启动 FastAPI 应用,绑定到所有 IP 地址 + uvicorn.run(app, host="0.0.0.0", port=8000) \ No newline at end of file diff --git a/AI/AiService/TestZhongHe.py b/AI/AiService/Test/CallDeepSeekPlain.py similarity index 70% rename from AI/AiService/TestZhongHe.py rename to AI/AiService/Test/CallDeepSeekPlain.py index 90ee4a60..a1c3b7b1 100644 --- a/AI/AiService/TestZhongHe.py +++ b/AI/AiService/Test/CallDeepSeekPlain.py @@ -6,7 +6,7 @@ from openai import OpenAI app = FastAPI() -# 阿里云中用来调用 deepseek r1 的密钥 +# 阿里云中用来调用 deepseek v3 的密钥 MODEL_API_KEY = "sk-01d13a39e09844038322108ecdbd1bbc" MODEL_NAME = "deepseek-v3" @@ -16,6 +16,7 @@ client = OpenAI( base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", ) + # 获取本机所有 IPv4 地址 def get_local_ips(): ips = [] @@ -31,27 +32,38 @@ def get_local_ips(): print(f"获取 IP 地址失败: {e}") return ips + # 模拟逐字生成数据的函数(SSE 格式) def generate_data(text): for char in text: time.sleep(0.05) # 模拟延迟 - #yield f"data: {char}\n\n".encode("utf-8") # 按照 SSE 格式返回数据 + # yield f"data: {char}\n\n".encode("utf-8") # 按照 SSE 格式返回数据 yield f"{char}".encode("utf-8") + + # 根路由,返回提示信息 @app.get("/") def root(): return PlainTextResponse("Hello ApiStream") + # 流式返回数据(SSE) +''' +http://10.10.21.20:8000/stream?course_name=%E9%97%B0%E5%9C%9F +''' @app.get("/stream") -def stream_data(): +def stream_data(course_name: str): + if not course_name: + return PlainTextResponse("请提供课程名称,例:course_name=三角形面积") # 调用阿里云 API 获取文本内容 completion = client.chat.completions.create( model=MODEL_NAME, messages=[ - {'role': 'system', 'content': 'You are a helpful assistant.'}, - {'role': 'user', 'content': '你是谁?'} + {'role': 'system', 'content': '你是一个教学经验丰富的基础教育教师'}, + {'role': 'user', 'content': '帮我设计一下' + course_name + '的课件提纲,用markdown格式返回。'} ], + timeout=6000, + # stream=True, ) text_content = completion.choices[0].message.content @@ -61,12 +73,13 @@ def stream_data(): media_type="text/event-stream", # 使用 text/event-stream headers={ "Cache-Control": "no-cache", # 禁用缓存 - "Connection": "keep-alive", # 保持连接 + "Connection": "keep-alive", # 保持连接 "Access-Control-Allow-Origin": "*", # 允许跨域 - "X-Accel-Buffering": "no" # 禁用 Nginx 缓冲(如果有代理) + "X-Accel-Buffering": "no" # 禁用 Nginx 缓冲(如果有代理) } ) + # 运行应用 if __name__ == "__main__": import uvicorn @@ -83,4 +96,4 @@ if __name__ == "__main__": print(f"http://{ip}:8000") # 启动 FastAPI 应用,绑定到所有 IP 地址 - uvicorn.run(app, host="0.0.0.0", port=8000) \ No newline at end of file + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/AI/AiService/TestChat.py b/AI/AiService/Test/TestChat.py similarity index 100% rename from AI/AiService/TestChat.py rename to AI/AiService/Test/TestChat.py diff --git a/AI/AiService/TestStream.py b/AI/AiService/Test/TestStream.py similarity index 100% rename from AI/AiService/TestStream.py rename to AI/AiService/Test/TestStream.py