481 lines
17 KiB
Python
481 lines
17 KiB
Python
|
"""
|
|||
|
|
DeepSeek 健康评估与建议生成服务
|
|||
|
|
用于生成"整体健康状况"和"功能性健康建议"内容
|
|||
|
|
|
|||
|
|
优化：优先使用模板中已有的项目解释，只有模板中没有的项目才调用 DeepSeek 生成
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import os
|
|||
|
|
import json
|
|||
|
|
import requests
|
|||
|
|
from pathlib import Path
|
|||
|
|
from typing import List, Dict, Any
|
|||
|
|
|
|||
|
|
|
|||
|
|
class DeepSeekHealthService:
|
|||
|
|
"""DeepSeek 健康内容生成服务"""
|
|||
|
|
|
|||
|
|
def __init__(self, api_key: str = None):
|
|||
|
|
self.api_key = api_key or os.getenv("DEEPSEEK_API_KEY", "")
|
|||
|
|
self.api_url = "https://api.deepseek.com/v1/chat/completions"
|
|||
|
|
|
|||
|
|
# 加载模板中的解释
|
|||
|
|
self.template_explanations = self._load_template_explanations()
|
|||
|
|
|
|||
|
|
def _load_template_explanations(self) -> Dict[str, Dict[str, str]]:
|
|||
|
|
"""加载模板中已有的项目解释"""
|
|||
|
|
explanations_file = Path(__file__).parent.parent / "template_explanations.json"
|
|||
|
|
|
|||
|
|
if explanations_file.exists():
|
|||
|
|
try:
|
|||
|
|
with open(explanations_file, 'r', encoding='utf-8') as f:
|
|||
|
|
explanations = json.load(f)
|
|||
|
|
print(f" ✓ 已加载 {len(explanations)} 个模板解释")
|
|||
|
|
return explanations
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" ⚠️ 加载模板解释失败: {e}")
|
|||
|
|
|
|||
|
|
return {}
|
|||
|
|
|
|||
|
|
def get_template_explanation(self, abb: str) -> Dict[str, str]:
|
|||
|
|
"""
|
|||
|
|
获取模板中的项目解释
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
abb: 项目缩写
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
{"clinical_en": "...", "clinical_cn": "..."} 或空字典
|
|||
|
|
"""
|
|||
|
|
# 尝试多种匹配方式
|
|||
|
|
abb_upper = abb.upper().strip()
|
|||
|
|
|
|||
|
|
# 直接匹配
|
|||
|
|
if abb_upper in self.template_explanations:
|
|||
|
|
return self.template_explanations[abb_upper]
|
|||
|
|
|
|||
|
|
# 去除特殊字符后匹配
|
|||
|
|
abb_clean = ''.join(c for c in abb_upper if c.isalnum())
|
|||
|
|
for key, value in self.template_explanations.items():
|
|||
|
|
key_clean = ''.join(c for c in key if c.isalnum())
|
|||
|
|
if abb_clean == key_clean:
|
|||
|
|
return value
|
|||
|
|
|
|||
|
|
return {}
|
|||
|
|
|
|||
|
|
def is_available(self) -> bool:
|
|||
|
|
"""检查服务是否可用"""
|
|||
|
|
return bool(self.api_key)
|
|||
|
|
|
|||
|
|
def call_deepseek(self, prompt: str) -> str:
|
|||
|
|
"""调用 DeepSeek API"""
|
|||
|
|
if not self.api_key:
|
|||
|
|
raise ValueError("未配置 DEEPSEEK_API_KEY")
|
|||
|
|
|
|||
|
|
headers = {
|
|||
|
|
"Authorization": f"Bearer {self.api_key}",
|
|||
|
|
"Content-Type": "application/json"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
data = {
|
|||
|
|
"model": "deepseek-chat",
|
|||
|
|
"messages": [{"role": "user", "content": prompt}],
|
|||
|
|
"temperature": 0.1,
|
|||
|
|
"max_tokens": 8000
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
response = requests.post(
|
|||
|
|
self.api_url,
|
|||
|
|
headers=headers,
|
|||
|
|
json=data,
|
|||
|
|
timeout=120
|
|||
|
|
)
|
|||
|
|
response.raise_for_status()
|
|||
|
|
return response.json()["choices"][0]["message"]["content"]
|
|||
|
|
|
|||
|
|
def collect_abnormal_items(self, analysis: Dict[str, Any]) -> List[Dict[str, str]]:
|
|||
|
|
"""
|
|||
|
|
从分析结果中收集异常项
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
analysis: LLM 分析结果，包含 abnormal_items 字段
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
异常项列表
|
|||
|
|
"""
|
|||
|
|
abnormal_items = []
|
|||
|
|
|
|||
|
|
# 从 abnormal_items 字段提取
|
|||
|
|
raw_items = analysis.get("abnormal_items", [])
|
|||
|
|
|
|||
|
|
for item in raw_items:
|
|||
|
|
if isinstance(item, dict):
|
|||
|
|
abnormal_items.append({
|
|||
|
|
"name": item.get("name", ""),
|
|||
|
|
"abb": item.get("abb", item.get("name", "")),
|
|||
|
|
"result": str(item.get("result", item.get("value", ""))),
|
|||
|
|
"reference": item.get("reference", ""),
|
|||
|
|
"unit": item.get("unit", ""),
|
|||
|
|
"level": item.get("level", ""),
|
|||
|
|
"point": "↑" if item.get("level") == "high" else ("↓" if item.get("level") == "low" else "")
|
|||
|
|
})
|
|||
|
|
elif isinstance(item, str):
|
|||
|
|
# 如果是字符串格式，尝试解析
|
|||
|
|
abnormal_items.append({
|
|||
|
|
"name": item,
|
|||
|
|
"abb": "",
|
|||
|
|
"result": "",
|
|||
|
|
"reference": "",
|
|||
|
|
"unit": "",
|
|||
|
|
"level": "",
|
|||
|
|
"point": ""
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
return abnormal_items
|
|||
|
|
|
|||
|
|
def get_item_explanations(self, abnormal_items: List[Dict[str, str]]) -> Dict[str, Dict[str, str]]:
|
|||
|
|
"""
|
|||
|
|
为异常项获取解释（优先使用模板中的解释，缺失的才调用 DeepSeek 生成）
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
abnormal_items: 异常项列表
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
{
|
|||
|
|
"ABB": {"clinical_en": "...", "clinical_cn": "..."},
|
|||
|
|
...
|
|||
|
|
}
|
|||
|
|
"""
|
|||
|
|
explanations = {}
|
|||
|
|
items_need_generation = []
|
|||
|
|
|
|||
|
|
print("\n 📋 检查模板中的项目解释...")
|
|||
|
|
|
|||
|
|
for item in abnormal_items:
|
|||
|
|
abb = item.get("abb", "").upper().strip()
|
|||
|
|
name = item.get("name", "")
|
|||
|
|
|
|||
|
|
if not abb:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
# 尝试从模板获取解释
|
|||
|
|
template_exp = self.get_template_explanation(abb)
|
|||
|
|
|
|||
|
|
if template_exp and template_exp.get("clinical_en") and template_exp.get("clinical_cn"):
|
|||
|
|
explanations[abb] = template_exp
|
|||
|
|
print(f" ✓ {abb}: 使用模板解释")
|
|||
|
|
else:
|
|||
|
|
items_need_generation.append(item)
|
|||
|
|
print(f" ○ {abb}: 需要生成解释")
|
|||
|
|
|
|||
|
|
# 如果有需要生成的项目，调用 DeepSeek
|
|||
|
|
if items_need_generation and self.api_key:
|
|||
|
|
print(f"\n 🤖 调用 DeepSeek 为 {len(items_need_generation)} 个项目生成解释...")
|
|||
|
|
generated = self._generate_missing_explanations(items_need_generation)
|
|||
|
|
explanations.update(generated)
|
|||
|
|
|
|||
|
|
return explanations
|
|||
|
|
|
|||
|
|
def _generate_missing_explanations(self, items: List[Dict[str, str]]) -> Dict[str, Dict[str, str]]:
|
|||
|
|
"""
|
|||
|
|
调用 DeepSeek 为缺失解释的项目生成临床意义
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
items: 需要生成解释的项目列表
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
生成的解释字典
|
|||
|
|
"""
|
|||
|
|
if not items:
|
|||
|
|
return {}
|
|||
|
|
|
|||
|
|
# 构建项目描述
|
|||
|
|
items_desc = []
|
|||
|
|
for item in items:
|
|||
|
|
desc = f"- {item['abb']}: {item['name']}"
|
|||
|
|
if item.get('result'):
|
|||
|
|
desc += f", 结果: {item['result']}"
|
|||
|
|
if item.get('unit'):
|
|||
|
|
desc += f" {item['unit']}"
|
|||
|
|
if item.get('reference'):
|
|||
|
|
desc += f", 参考范围: {item['reference']}"
|
|||
|
|
items_desc.append(desc)
|
|||
|
|
|
|||
|
|
prompt = f"""你是一位医学检验专家，请为以下医疗检测项目生成临床意义解释。
|
|||
|
|
|
|||
|
|
## 需要解释的项目：
|
|||
|
|
{chr(10).join(items_desc)}
|
|||
|
|
|
|||
|
|
## 要求：
|
|||
|
|
1. 为每个项目提供英文和中文的临床意义解释
|
|||
|
|
2. 解释应包含：该指标的作用、正常范围的意义、异常时可能的原因
|
|||
|
|
3. 语言专业但易于理解
|
|||
|
|
4. 每个解释约50-100字
|
|||
|
|
|
|||
|
|
## 输出格式（JSON）：
|
|||
|
|
```json
|
|||
|
|
{{
|
|||
|
|
"ABB1": {{
|
|||
|
|
"clinical_en": "English clinical significance...",
|
|||
|
|
"clinical_cn": "中文临床意义..."
|
|||
|
|
}},
|
|||
|
|
"ABB2": {{
|
|||
|
|
"clinical_en": "...",
|
|||
|
|
"clinical_cn": "..."
|
|||
|
|
}}
|
|||
|
|
}}
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
只返回JSON，不要其他说明。"""
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
response = self.call_deepseek(prompt)
|
|||
|
|
|
|||
|
|
# 解析 JSON
|
|||
|
|
if "```json" in response:
|
|||
|
|
response = response.split("```json")[1].split("```")[0]
|
|||
|
|
elif "```" in response:
|
|||
|
|
response = response.split("```")[1].split("```")[0]
|
|||
|
|
|
|||
|
|
result = json.loads(response.strip())
|
|||
|
|
print(f" ✓ 成功生成 {len(result)} 个项目的解释")
|
|||
|
|
return result
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" ✗ 生成解释失败: {e}")
|
|||
|
|
return {}
|
|||
|
|
|
|||
|
|
def generate_health_assessment(self, abnormal_items: List[Dict[str, str]]) -> Dict[str, Any]:
|
|||
|
|
"""
|
|||
|
|
生成"整体健康状况"评估内容
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
abnormal_items: 异常项列表
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
包含多个小节的健康评估内容
|
|||
|
|
"""
|
|||
|
|
if not self.api_key or not abnormal_items:
|
|||
|
|
return {"sections": []}
|
|||
|
|
|
|||
|
|
# 构建异常项描述
|
|||
|
|
abnormal_desc = []
|
|||
|
|
for item in abnormal_items:
|
|||
|
|
direction = "偏高" if item.get("point") in ["↑", "H", "高"] or item.get("level") == "high" else "偏低"
|
|||
|
|
desc = f"- {item['name']}"
|
|||
|
|
if item.get('abb'):
|
|||
|
|
desc += f" ({item['abb']})"
|
|||
|
|
desc += f": {item['result']}"
|
|||
|
|
if item.get('unit'):
|
|||
|
|
desc += f" {item['unit']}"
|
|||
|
|
desc += f" ({direction}"
|
|||
|
|
if item.get('reference'):
|
|||
|
|
desc += f", 参考范围: {item['reference']}"
|
|||
|
|
desc += ")"
|
|||
|
|
abnormal_desc.append(desc)
|
|||
|
|
|
|||
|
|
prompt = f"""你是一位功能医学专家，请根据以下所有异常检测指标，撰写"整体健康状况评估"的内容。
|
|||
|
|
|
|||
|
|
## 异常指标：
|
|||
|
|
{chr(10).join(abnormal_desc)}
|
|||
|
|
|
|||
|
|
## 要求：
|
|||
|
|
1. 根据异常指标的类型，自动分成合适的小节（如血液学、内分泌、免疫、代谢等，根据实际异常项决定）
|
|||
|
|
2. 每个小节包含英文和中文两个版本
|
|||
|
|
3. 从功能医学和整体健康角度分析
|
|||
|
|
4. 解释可能的原因和健康影响
|
|||
|
|
5. 语言专业但易于理解
|
|||
|
|
6. 每个小节的每个语言版本约150-250字
|
|||
|
|
|
|||
|
|
## 输出格式（JSON）：
|
|||
|
|
```json
|
|||
|
|
{{
|
|||
|
|
"sections": [
|
|||
|
|
{{
|
|||
|
|
"title_en": "(I) Section Title in English",
|
|||
|
|
"title_cn": "（一）中文小节标题",
|
|||
|
|
"content_en": "English analysis content...",
|
|||
|
|
"content_cn": "中文分析内容..."
|
|||
|
|
}}
|
|||
|
|
]
|
|||
|
|
}}
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
只返回JSON，不要其他说明。根据实际异常项情况决定分几个小节，不要硬套固定模板。"""
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
response = self.call_deepseek(prompt)
|
|||
|
|
|
|||
|
|
# 解析 JSON
|
|||
|
|
if "```json" in response:
|
|||
|
|
response = response.split("```json")[1].split("```")[0]
|
|||
|
|
elif "```" in response:
|
|||
|
|
response = response.split("```")[1].split("```")[0]
|
|||
|
|
|
|||
|
|
result = json.loads(response.strip())
|
|||
|
|
print(f" ✓ 生成健康评估内容，共 {len(result.get('sections', []))} 个小节")
|
|||
|
|
return result
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" ✗ 生成健康评估内容失败: {e}")
|
|||
|
|
return {"sections": []}
|
|||
|
|
|
|||
|
|
def generate_health_advice(self, abnormal_items: List[Dict[str, str]]) -> Dict[str, Any]:
|
|||
|
|
"""
|
|||
|
|
生成"功能性健康建议"内容
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
abnormal_items: 异常项列表
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
包含5个固定小节的健康建议内容
|
|||
|
|
"""
|
|||
|
|
if not self.api_key or not abnormal_items:
|
|||
|
|
return {"sections": []}
|
|||
|
|
|
|||
|
|
# 异常项描述
|
|||
|
|
abnormal_desc = []
|
|||
|
|
for item in abnormal_items:
|
|||
|
|
direction = "偏高" if item.get("point") in ["↑", "H", "高"] or item.get("level") == "high" else "偏低"
|
|||
|
|
desc = f"- {item['name']}"
|
|||
|
|
if item.get('abb'):
|
|||
|
|
desc += f" ({item['abb']})"
|
|||
|
|
desc += f": {item['result']}"
|
|||
|
|
if item.get('unit'):
|
|||
|
|
desc += f" {item['unit']}"
|
|||
|
|
desc += f" ({direction})"
|
|||
|
|
abnormal_desc.append(desc)
|
|||
|
|
|
|||
|
|
prompt = f"""你是一位功能医学专家，请根据以下异常检测指标，撰写"功能医学健康建议"的内容。
|
|||
|
|
|
|||
|
|
## 异常指标：
|
|||
|
|
{chr(10).join(abnormal_desc)}
|
|||
|
|
|
|||
|
|
## 要求：
|
|||
|
|
1. 必须包含以下5个固定小节（按顺序）：
|
|||
|
|
- Nutrition Intervention 营养干预
|
|||
|
|
- Exercise Intervention 运动干预
|
|||
|
|
- Sleep & Stress Management 睡眠与压力管理
|
|||
|
|
- Lifestyle Adjustment 生活方式调整
|
|||
|
|
- Long-term Follow-up Plan 长期随访计划
|
|||
|
|
2. 每个小节针对这些异常指标提供具体、可执行的建议
|
|||
|
|
3. 从功能医学角度出发，强调预防和整体调理
|
|||
|
|
4. 每个小节包含3-5条具体建议措施
|
|||
|
|
5. 语言专业但易于理解
|
|||
|
|
6. 分别提供英文和中文版本
|
|||
|
|
7. 每个小节的每个语言版本约200-300字
|
|||
|
|
|
|||
|
|
## 输出格式（JSON）：
|
|||
|
|
```json
|
|||
|
|
{{
|
|||
|
|
"sections": [
|
|||
|
|
{{
|
|||
|
|
"title_en": "Nutrition Intervention",
|
|||
|
|
"title_cn": "营养干预",
|
|||
|
|
"content_en": "English nutrition advice...",
|
|||
|
|
"content_cn": "中文营养建议..."
|
|||
|
|
}},
|
|||
|
|
{{
|
|||
|
|
"title_en": "Exercise Intervention",
|
|||
|
|
"title_cn": "运动干预",
|
|||
|
|
"content_en": "...",
|
|||
|
|
"content_cn": "..."
|
|||
|
|
}},
|
|||
|
|
{{
|
|||
|
|
"title_en": "Sleep & Stress Management",
|
|||
|
|
"title_cn": "睡眠与压力管理",
|
|||
|
|
"content_en": "...",
|
|||
|
|
"content_cn": "..."
|
|||
|
|
}},
|
|||
|
|
{{
|
|||
|
|
"title_en": "Lifestyle Adjustment",
|
|||
|
|
"title_cn": "生活方式调整",
|
|||
|
|
"content_en": "...",
|
|||
|
|
"content_cn": "..."
|
|||
|
|
}},
|
|||
|
|
{{
|
|||
|
|
"title_en": "Long-term Follow-up Plan",
|
|||
|
|
"title_cn": "长期随访计划",
|
|||
|
|
"content_en": "...",
|
|||
|
|
"content_cn": "..."
|
|||
|
|
}}
|
|||
|
|
]
|
|||
|
|
}}
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
只返回JSON，不要其他说明。"""
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
response = self.call_deepseek(prompt)
|
|||
|
|
|
|||
|
|
# 解析 JSON
|
|||
|
|
if "```json" in response:
|
|||
|
|
response = response.split("```json")[1].split("```")[0]
|
|||
|
|
elif "```" in response:
|
|||
|
|
response = response.split("```")[1].split("```")[0]
|
|||
|
|
|
|||
|
|
result = json.loads(response.strip())
|
|||
|
|
print(f" ✓ 生成健康建议内容，共 {len(result.get('sections', []))} 个小节")
|
|||
|
|
return result
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" ✗ 生成健康建议内容失败: {e}")
|
|||
|
|
return {"sections": []}
|
|||
|
|
|
|||
|
|
def generate_health_content(self, analysis: Dict[str, Any]) -> Dict[str, Any]:
|
|||
|
|
"""
|
|||
|
|
生成完整的健康评估和建议内容
|
|||
|
|
|
|||
|
|
优化：优先使用模板中已有的项目解释，只有模板中没有的项目才调用 DeepSeek 生成
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
analysis: LLM 分析结果
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
包含 health_assessment, health_advice, item_explanations 的字典
|
|||
|
|
"""
|
|||
|
|
if not self.is_available():
|
|||
|
|
print(" ⚠️ DeepSeek API Key 未配置，跳过健康内容生成")
|
|||
|
|
return {}
|
|||
|
|
|
|||
|
|
print("\n============================================================")
|
|||
|
|
print("DeepSeek 健康内容生成")
|
|||
|
|
print("============================================================")
|
|||
|
|
|
|||
|
|
# 收集异常项
|
|||
|
|
print("\n 📝 正在收集异常项...")
|
|||
|
|
abnormal_items = self.collect_abnormal_items(analysis)
|
|||
|
|
|
|||
|
|
if not abnormal_items:
|
|||
|
|
print(" ℹ️ 没有检测到异常项目，跳过内容生成")
|
|||
|
|
return {}
|
|||
|
|
|
|||
|
|
print(f" 发现 {len(abnormal_items)} 个异常项目:")
|
|||
|
|
for item in abnormal_items[:10]:
|
|||
|
|
direction = "↑" if item.get("level") == "high" or item.get("point") == "↑" else "↓"
|
|||
|
|
print(f" - {item['name']}: {item['result']} {direction}")
|
|||
|
|
if len(abnormal_items) > 10:
|
|||
|
|
print(f" ... 等共 {len(abnormal_items)} 项")
|
|||
|
|
|
|||
|
|
# 获取项目解释（优先使用模板，缺失的才生成）
|
|||
|
|
item_explanations = self.get_item_explanations(abnormal_items)
|
|||
|
|
|
|||
|
|
# 统计使用情况
|
|||
|
|
template_count = sum(1 for abb in item_explanations if self.get_template_explanation(abb))
|
|||
|
|
generated_count = len(item_explanations) - template_count
|
|||
|
|
print(f"\n 📊 解释来源统计: 模板 {template_count} 个, DeepSeek生成 {generated_count} 个")
|
|||
|
|
|
|||
|
|
# 生成健康评估
|
|||
|
|
print("\n 🤖 正在调用 DeepSeek 生成整体健康状况...")
|
|||
|
|
health_assessment = self.generate_health_assessment(abnormal_items)
|
|||
|
|
|
|||
|
|
# 生成健康建议
|
|||
|
|
print("\n 🤖 正在调用 DeepSeek 生成功能性健康建议...")
|
|||
|
|
health_advice = self.generate_health_advice(abnormal_items)
|
|||
|
|
|
|||
|
|
print("\n ✓ 健康内容生成完成")
|
|||
|
|
|
|||
|
|
return {
|
|||
|
|
"health_assessment": health_assessment,
|
|||
|
|
"health_advice": health_advice,
|
|||
|
|
"abnormal_items": abnormal_items,
|
|||
|
|
"item_explanations": item_explanations # 包含每个项目的解释
|
|||
|
|
}
|