|
import os |
|
import json |
|
import gradio as gr |
|
import requests |
|
from dotenv import load_dotenv |
|
|
|
|
|
load_dotenv() |
|
API_KEY = os.getenv("DEEPSEEK_API_KEY") |
|
API_KEY = "sk-0d040569ee2b4932a807d56a5b12ac05" |
|
API_URL = "https://api.deepseek.com/v1/chat/completions" |
|
|
|
def build_messages(query: str, history: list): |
|
"""构造对话历史""" |
|
messages = [{"role": "system", "content": "你是一个有帮助的助手"}] |
|
for user_msg, bot_msg in history: |
|
messages.extend([ |
|
{"role": "user", "content": user_msg}, |
|
{"role": "assistant", "content": bot_msg} |
|
]) |
|
messages.append({"role": "user", "content": query}) |
|
return messages |
|
|
|
def stream_response(query: str, history: list): |
|
"""流式响应生成器""" |
|
headers = { |
|
"Authorization": f"Bearer {API_KEY}", |
|
"Content-Type": "application/json" |
|
} |
|
|
|
try: |
|
response = requests.post( |
|
API_URL, |
|
headers=headers, |
|
json={ |
|
"model": "deepseek-chat", |
|
"messages": build_messages(query, history), |
|
"temperature": 0.7, |
|
"stream": True |
|
}, |
|
stream=True, |
|
timeout=60 |
|
) |
|
|
|
|
|
if response.status_code != 200: |
|
error_msg = f"API 返回错误状态码: {response.status_code} - {response.text}" |
|
print(error_msg) |
|
yield error_msg |
|
return |
|
|
|
partial_message = "" |
|
for chunk in response.iter_lines(): |
|
|
|
if not chunk or b'[DONE]' in chunk: |
|
continue |
|
|
|
try: |
|
|
|
print("原始 chunk:", chunk) |
|
|
|
decoded = chunk.decode('utf-8').strip() |
|
|
|
if decoded.startswith('data:'): |
|
json_str = decoded[5:].strip() |
|
else: |
|
json_str = decoded |
|
|
|
|
|
if not json_str.startswith('{'): |
|
continue |
|
|
|
data = json.loads(json_str) |
|
|
|
if content := data['choices'][0]['delta'].get('content'): |
|
partial_message += content |
|
yield partial_message |
|
|
|
except json.JSONDecodeError as e: |
|
print(f"JSON 解析失败: {e} | 原始数据: {decoded}") |
|
continue |
|
except Exception as e: |
|
print(f"处理 chunk 时发生错误: {str(e)}") |
|
continue |
|
|
|
except Exception as e: |
|
yield f"⚠️ 请求失败:{str(e)}" |
|
|
|
def stream_response_back(query: str, history: list): |
|
"""流式响应生成器""" |
|
headers = { |
|
"Authorization": f"Bearer {API_KEY}", |
|
"Content-Type": "application/json" |
|
} |
|
|
|
print(query) |
|
print(history) |
|
|
|
try: |
|
response = requests.post( |
|
API_URL, |
|
headers=headers, |
|
json={ |
|
"model": "deepseek-chat", |
|
"messages": build_messages(query, history), |
|
"temperature": 0.7, |
|
"stream": True |
|
}, |
|
stream=True, |
|
timeout=60 |
|
) |
|
response.raise_for_status() |
|
|
|
print(response) |
|
|
|
partial_message = "" |
|
for chunk in response.iter_lines(): |
|
if chunk: |
|
|
|
decoded = chunk.decode('utf-8').strip() |
|
if decoded.startswith('data:'): |
|
data = json.loads(decoded[5:]) |
|
if content := data['choices'][0]['delta'].get('content'): |
|
partial_message += content |
|
yield partial_message |
|
|
|
except Exception as e: |
|
yield f"⚠️ 请求失败:{str(e)}" |
|
|
|
|
|
demo = gr.ChatInterface( |
|
fn=stream_response, |
|
title="DeepSeek 智能助手", |
|
description="输入消息开始对话(支持流式打字效果)", |
|
theme="soft", |
|
examples=["你好!", "如何学习AI?", "写一首关于春天的诗"], |
|
cache_examples=False, |
|
|
|
|
|
|
|
stop_btn="停止生成", |
|
) |
|
|
|
if __name__ == "__main__": |
|
demo.launch(server_name="0.0.0.0", server_port=7860) |