217 lines
6.5 KiB
Python
217 lines
6.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
CardioAI - Module 3: AI 语音助手
|
|
Deepseek 问答 + CosyVoice 语音合成服务
|
|
"""
|
|
|
|
import os
|
|
import base64
|
|
import dashscope
|
|
from pathlib import Path
|
|
from flask import Flask, render_template, request, jsonify
|
|
from dotenv import load_dotenv
|
|
from langchain_openai import ChatOpenAI
|
|
from dashscope.audio.tts_v2 import SpeechSynthesizer, AudioFormat
|
|
|
|
# ============================================
|
|
# 配置与初始化
|
|
# ============================================
|
|
# 加载环境变量
|
|
ENV_PATH = Path(r"E:\project_ai\claude_project1\aicodes\.env")
|
|
load_dotenv(dotenv_path=ENV_PATH)
|
|
|
|
# 配置 DashScope API Key
|
|
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")
|
|
|
|
# 创建 Flask 应用
|
|
CODE_ROOT = Path(r"E:\project_ai\claude_project1\aicodes")
|
|
TEMPLATE_DIR = CODE_ROOT / "module3_voice_assistant" / "templates"
|
|
|
|
app = Flask(__name__,
|
|
template_folder=str(TEMPLATE_DIR),
|
|
static_folder=str(TEMPLATE_DIR.parent / 'static'))
|
|
|
|
|
|
# ============================================
|
|
# DeepSeek LLM 配置
|
|
# ============================================
|
|
def get_llm():
|
|
"""
|
|
初始化 DeepSeek LLM 实例
|
|
|
|
Returns:
|
|
ChatOpenAI: DeepSeek LLM 实例
|
|
"""
|
|
llm = ChatOpenAI(
|
|
base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
|
|
api_key=os.getenv("DEEPSEEK_API_KEY"),
|
|
model=os.getenv("MODEL_NAME", "deepseek-chat"),
|
|
temperature=float(os.getenv("TEMPERATURE", 0.7)),
|
|
max_tokens=int(os.getenv("MAX_TOKENS", 2000))
|
|
)
|
|
return llm
|
|
|
|
|
|
# System Prompt - 专业心血管健康顾问
|
|
SYSTEM_PROMPT = """你是一位专业的心血管健康顾问，名为 CardioAI 助手。
|
|
|
|
你的职责：
|
|
1. 回答用户关于心血管疾病预防、诊断、治疗和管理的问题
|
|
2. 提供基于医学证据的健康建议
|
|
3. 解释心血管相关的医学概念和检查指标
|
|
4. 给出生活方式改善建议（饮食、运动、戒烟限酒等）
|
|
|
|
注意事项：
|
|
- 回答要专业、准确、通俗易懂
|
|
- 对于严重的健康问题，建议用户及时就医
|
|
- 不要替代医生的诊断，你的建议仅供参考
|
|
- 回答要简洁明了，避免过于冗长
|
|
|
|
请用温暖、专业的语气回答用户的问题。"""
|
|
|
|
|
|
# ============================================
|
|
# CosyVoice 语音合成函数
|
|
# ============================================
|
|
def text_to_speech(text: str) -> bytes:
|
|
"""
|
|
使用 CosyVoice 将文本转换为语音
|
|
|
|
Args:
|
|
text (str): 待合成的文本
|
|
|
|
Returns:
|
|
bytes: MP3 格式的音频数据
|
|
"""
|
|
try:
|
|
# 实例化 SpeechSynthesizer
|
|
synthesizer = SpeechSynthesizer(
|
|
model=os.getenv("VOICE_MODEL", "cosyvoice-v2"),
|
|
voice=os.getenv("VOICE", "longxiaochun_v2"),
|
|
format=AudioFormat.MP3_22050HZ_MONO_256KBPS
|
|
)
|
|
|
|
# 同步调用 - 将文本转换为语音
|
|
audio_data = synthesizer.call(text)
|
|
|
|
return audio_data
|
|
|
|
except Exception as e:
|
|
print(f"语音合成失败: {str(e)}")
|
|
return None
|
|
|
|
|
|
# ============================================
|
|
# 路由定义
|
|
# ============================================
|
|
@app.route('/')
|
|
def index():
|
|
"""主页 - 渲染语音助手界面"""
|
|
return render_template('voice_index.html')
|
|
|
|
|
|
@app.route('/voice_assistant', methods=['POST'])
|
|
def voice_assistant():
|
|
"""
|
|
AI 语音助手 API 接口
|
|
|
|
接收用户问题，返回 LLM 文字回答和 Base64 编码的语音
|
|
"""
|
|
try:
|
|
# 获取请求数据
|
|
data = request.get_json()
|
|
user_question = data.get('question', '').strip()
|
|
|
|
if not user_question:
|
|
return jsonify({
|
|
'error': '请输入您的问题'
|
|
}), 400
|
|
|
|
# 检查 API Key 配置
|
|
deepseek_key = os.getenv("DEEPSEEK_API_KEY")
|
|
dashscope_key = os.getenv("DASHSCOPE_API_KEY")
|
|
|
|
if not deepseek_key or deepseek_key == "your_deepseek_api_key_here":
|
|
return jsonify({
|
|
'error': 'DeepSeek API Key 未配置，请在 .env 文件中设置 DEEPSEEK_API_KEY'
|
|
}), 500
|
|
|
|
if not dashscope_key or dashscope_key == "your_dashscope_api_key_here":
|
|
return jsonify({
|
|
'error': 'DashScope API Key 未配置，请在 .env 文件中设置 DASHSCOPE_API_KEY'
|
|
}), 500
|
|
|
|
print(f"\n用户问题: {user_question}")
|
|
|
|
# 1. 调用 DeepSeek LLM 获取文字回答
|
|
print("正在调用 DeepSeek LLM...")
|
|
llm = get_llm()
|
|
|
|
# 构建完整 prompt
|
|
messages = [
|
|
{"role": "system", "content": SYSTEM_PROMPT},
|
|
{"role": "user", "content": user_question}
|
|
]
|
|
|
|
# 调用 LLM
|
|
llm_response = llm.invoke(messages)
|
|
text_answer = llm_response.content
|
|
|
|
print(f"LLM 回答: {text_answer}")
|
|
|
|
# 2. 将文字回答转换为语音
|
|
print("正在调用 CosyVoice 合成语音...")
|
|
audio_data = text_to_speech(text_answer)
|
|
|
|
if audio_data is None:
|
|
return jsonify({
|
|
'error': '语音合成失败，请检查 API Key 配置和网络连接'
|
|
}), 500
|
|
|
|
# 3. 将音频编码为 Base64
|
|
audio_base64 = base64.b64encode(audio_data).decode('utf-8')
|
|
|
|
print(f"语音合成完成，音频大小: {len(audio_data)} 字节")
|
|
|
|
# 4. 返回结果
|
|
return jsonify({
|
|
'text_answer': text_answer,
|
|
'audio_base64': audio_base64,
|
|
'message': '问答成功'
|
|
})
|
|
|
|
except Exception as e:
|
|
print(f"处理失败: {str(e)}")
|
|
return jsonify({
|
|
'error': f'处理失败: {str(e)}'
|
|
}), 500
|
|
|
|
|
|
@app.route('/health', methods=['GET'])
|
|
def health():
|
|
"""健康检查接口"""
|
|
deepseek_configured = os.getenv("DEEPSEEK_API_KEY") not in [None, "your_deepseek_api_key_here"]
|
|
dashscope_configured = os.getenv("DASHSCOPE_API_KEY") not in [None, "your_dashscope_api_key_here"]
|
|
|
|
return jsonify({
|
|
'status': 'healthy',
|
|
'deepseek_configured': deepseek_configured,
|
|
'dashscope_configured': dashscope_configured
|
|
})
|
|
|
|
|
|
# ============================================
|
|
# 主程序入口
|
|
# ============================================
|
|
if __name__ == '__main__':
|
|
host = os.getenv('FLASK_HOST', '127.0.0.1')
|
|
port = int(os.getenv('FLASK_PORT', 5001)) # 使用 5001 避免与 Module 2 冲突
|
|
|
|
print("\n" + "=" * 50)
|
|
print("CardioAI - Module 3: AI 语音助手服务")
|
|
print("=" * 50)
|
|
print(f"服务地址: http://{host}:{port}")
|
|
print("=" * 50 + "\n")
|
|
|
|
app.run(host=host, port=port, debug=True)
|