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 # 包含每个项目的解释
|
||
}
|