240 lines
8.0 KiB
Python
240 lines
8.0 KiB
Python
"""
|
||
AI service: Report analysis, chat, doctor questions.
|
||
Uses OpenAI-compatible API (e.g. SiliconFlow).
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
from typing import Optional
|
||
|
||
from openai import AsyncOpenAI
|
||
|
||
from app.config import settings
|
||
from app.models.report import Report
|
||
from app.models.user import User
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _get_client() -> AsyncOpenAI:
|
||
return AsyncOpenAI(api_key=settings.ai_api_key, base_url=settings.ai_base_url)
|
||
|
||
|
||
def _build_report_context(report: Report, user: User) -> str:
|
||
"""构建报告上下文给 AI"""
|
||
lines = [
|
||
f"孕妈信息: 年龄{user.age}岁",
|
||
]
|
||
if report.pregnancy_week:
|
||
lines.append(f"检查时孕周: 第{report.pregnancy_week}周")
|
||
if report.report_date:
|
||
lines.append(f"报告日期: {report.report_date}")
|
||
if user.last_period_date:
|
||
lines.append(f"末次月经: {user.last_period_date}")
|
||
if user.height:
|
||
lines.append(f"身高: {user.height}cm")
|
||
if user.current_weight:
|
||
lines.append(f"当前体重: {user.current_weight}kg")
|
||
if user.blood_type:
|
||
lines.append(f"血型: {user.blood_type}")
|
||
if user.medical_history:
|
||
lines.append(f"病史: {', '.join(user.medical_history)}")
|
||
|
||
lines.append(f"\n报告类型: {report.report_type}")
|
||
lines.append("\n检查指标:")
|
||
for item in report.data_items:
|
||
ref = f", 参考范围: {item.reference_range}" if item.reference_range else ""
|
||
lines.append(f" - {item.name}: {item.value} {item.unit}{ref}")
|
||
|
||
if report.symptoms:
|
||
lines.append(f"\n自觉症状: {', '.join(report.symptoms)}")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _parse_json_response(text: str) -> dict:
|
||
"""从 AI 回复中提取 JSON"""
|
||
if "```json" in text:
|
||
text = text.split("```json")[1].split("```")[0]
|
||
elif "```" in text:
|
||
text = text.split("```")[1].split("```")[0]
|
||
return json.loads(text.strip())
|
||
|
||
|
||
async def _chat_complete(system: str, user_content: str, messages: Optional[list] = None, max_tokens: int = 4096) -> str:
|
||
"""统一调用 AI 文本模型"""
|
||
client = _get_client()
|
||
msgs = [{"role": "system", "content": system}]
|
||
if messages:
|
||
msgs.extend(messages)
|
||
else:
|
||
msgs.append({"role": "user", "content": user_content})
|
||
|
||
try:
|
||
response = await client.chat.completions.create(
|
||
model=settings.ai_text_model,
|
||
max_tokens=max_tokens,
|
||
temperature=0.7, # 降低温度提高生成速度
|
||
messages=msgs,
|
||
)
|
||
except Exception as e:
|
||
err_msg = str(e)
|
||
logger.error("AI 调用失败: %s: %s", type(e).__name__, err_msg[:200])
|
||
if "auth" in err_msg.lower() or "api key" in err_msg.lower():
|
||
raise RuntimeError("AI 服务认证失败，请检查 API Key") from e
|
||
raise RuntimeError("AI 服务暂时不可用，请稍后重试") from e
|
||
|
||
return response.choices[0].message.content
|
||
|
||
|
||
# ─── Report Analysis ───
|
||
|
||
ANALYSIS_SYSTEM_PROMPT = """你是一位专业、温暖的孕期健康顾问AI。你的任务是解读孕期检查报告，用通俗易懂的语言帮助孕妈理解检查结果。
|
||
|
||
核心要求：
|
||
1. 必须结合孕周来判断指标是否正常（孕期不同阶段的正常范围不同，如血红蛋白在孕中晚期会生理性下降）
|
||
2. 将每个指标的测定结果与报告中的参考范围对比，准确判断偏高/偏低/正常
|
||
3. 如果参考范围缺失，根据你的医学知识和该孕周的标准范围来判断
|
||
4. 用温暖、关怀的语气，避免医学术语恐吓
|
||
5. 对每个指标给出白话化的昵称（如血红蛋白→"氧气快递员"，双顶径→"宝宝头围"，腹围→"宝宝肚围"，股骨长→"宝宝腿长"）
|
||
6. 解释每个指标的含义、对宝宝的影响、行动建议
|
||
7. 给出整体评估和风险等级
|
||
8. 行动建议按优先级排序
|
||
|
||
**特别注意：**
|
||
- 如果是B超报告，重点关注胎儿生长指标（双顶径、腹围、股骨长），即使只有这3个指标也要给出完整的解读
|
||
- B超报告的解读要重点说明宝宝的生长发育情况，是否符合孕周标准
|
||
- 即使指标数量很少（如只有3个），也要给出详细的解读和建议
|
||
|
||
请严格按以下JSON格式返回：
|
||
{
|
||
"overall_status": "normal/attention/warning",
|
||
"risk_level": "low/medium/high",
|
||
"overall_summary": "整体评估文案",
|
||
"emotional_summary": "情感化总结，如'宝宝发育良好，继续保持'",
|
||
"categories": [
|
||
{
|
||
"category": "分类名称",
|
||
"status": "normal/attention/warning",
|
||
"indicators": [
|
||
{
|
||
"name": "指标名",
|
||
"nickname": "白话昵称",
|
||
"value": "值",
|
||
"status": "normal/high/low",
|
||
"what_is_it": "通俗解释",
|
||
"impact": "对宝宝的影响",
|
||
"suggestion": "行动建议",
|
||
"recheck_reminder": "复查提醒（可选）"
|
||
}
|
||
]
|
||
}
|
||
],
|
||
"action_items": [
|
||
{
|
||
"content": "具体建议",
|
||
"priority": "high/medium/low",
|
||
"category": "饮食/运动/就医/复查"
|
||
}
|
||
],
|
||
"tips": ["温馨提示1", "温馨提示2"]
|
||
}"""
|
||
|
||
|
||
async def analyze_report(report: Report, user: User) -> dict:
|
||
"""调用 AI 解读报告"""
|
||
logger.info("开始AI解读报告 report_id=%s", report.id)
|
||
context = _build_report_context(report, user)
|
||
response_text = await _chat_complete(
|
||
system=ANALYSIS_SYSTEM_PROMPT,
|
||
user_content=f"请解读以下孕期检查报告:\n\n{context}",
|
||
)
|
||
try:
|
||
return _parse_json_response(response_text)
|
||
except (json.JSONDecodeError, IndexError):
|
||
return {
|
||
"overall_status": "normal",
|
||
"risk_level": "low",
|
||
"overall_summary": response_text,
|
||
"emotional_summary": "报告解读完成",
|
||
"categories": [],
|
||
"action_items": [],
|
||
"tips": [],
|
||
}
|
||
|
||
|
||
# ─── Chat ───
|
||
|
||
CHAT_SYSTEM_PROMPT = """你是一位专业、温暖的孕期健康顾问AI。孕妈正在就她的检查报告向你提问。
|
||
|
||
要求：
|
||
1. 用通俗易懂、温暖关怀的语气回答
|
||
2. 基于报告数据给出专业建议
|
||
3. 遇到严重问题建议及时就医
|
||
4. 回答简洁明了，避免过长
|
||
|
||
以下是孕妈的检查报告信息：
|
||
{report_context}"""
|
||
|
||
|
||
async def chat_with_ai(
|
||
report: Report, user: User, messages: list[dict], user_message: str
|
||
) -> str:
|
||
"""AI 对话追问"""
|
||
context = _build_report_context(report, user)
|
||
system = CHAT_SYSTEM_PROMPT.format(report_context=context)
|
||
|
||
conversation = [{"role": "system", "content": system}]
|
||
for msg in messages[-10:]:
|
||
conversation.append({"role": msg["role"], "content": msg["content"]})
|
||
conversation.append({"role": "user", "content": user_message})
|
||
|
||
client = _get_client()
|
||
try:
|
||
response = await client.chat.completions.create(
|
||
model=settings.ai_text_model,
|
||
max_tokens=1024,
|
||
messages=conversation,
|
||
)
|
||
except Exception as e:
|
||
logger.error("AI 对话失败: %s", e)
|
||
raise RuntimeError("AI 对话服务暂时不可用") from e
|
||
|
||
return response.choices[0].message.content
|
||
|
||
|
||
# ─── Doctor Questions ───
|
||
|
||
DOCTOR_QUESTIONS_SYSTEM_PROMPT = """你是一位孕期健康顾问AI。根据检查报告中的异常项，帮孕妈生成3-5条看医生时可以问的问题。
|
||
|
||
要求：
|
||
1. 问题要专业但容易理解
|
||
2. 按重要性排序
|
||
3. 给出为什么要问这个问题的原因
|
||
|
||
请严格按以下JSON格式返回：
|
||
{
|
||
"questions": [
|
||
{
|
||
"category": "分类",
|
||
"question": "问医生的话术",
|
||
"reason": "为什么要问这个",
|
||
"priority": "high/medium/low"
|
||
}
|
||
]
|
||
}"""
|
||
|
||
|
||
async def generate_doctor_questions(report: Report, user: User) -> dict:
|
||
"""生成问诊清单"""
|
||
context = _build_report_context(report, user)
|
||
response_text = await _chat_complete(
|
||
system=DOCTOR_QUESTIONS_SYSTEM_PROMPT,
|
||
user_content=f"请根据以下报告生成问诊清单:\n\n{context}",
|
||
max_tokens=1024,
|
||
)
|
||
try:
|
||
return _parse_json_response(response_text)
|
||
except (json.JSONDecodeError, IndexError):
|
||
return {"questions": []}
|