main
HuangHai 5 months ago
parent ffdb664760
commit 4abcda780f

@ -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)

@ -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)
uvicorn.run(app, host="0.0.0.0", port=8000)
Loading…
Cancel
Save