from fastapi import APIRouter, HTTPException from pydantic import BaseModel, ConfigDict from typing import Optional from app.graphs.chat_graph import create_chat_graph from app.graphs.analysis_graph import create_analysis_graph from app.graphs.health_coach_graph import create_health_coach_graph from app.graphs.health_butler_graph import create_health_butler_graph from app.graphs.nutrition_graph import create_nutrition_graph from app.tools.java_client import JavaClient from app.rag.retriever import RagRetriever from app.config import settings from app.prompt_service import get_prompt from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage from app.models.health_plan import HealthPlanResponse, PlanTask import uuid import time import re import logging import json logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1", tags=["adapter"]) # ---- Dify-compatible request/response models ---- class DifyMessage(BaseModel): role: str content: str class DifyChatRequest(BaseModel): # 前端 userId/conversationId 可能为数字，允许 int/float 自动转 str，避免 422 model_config = ConfigDict(coerce_numbers_to_str=True) query: str = "" user: str = "anonymous" conversation_id: str = "" messages: list[DifyMessage] = [] inputs: dict = {} response_mode: str = "blocking" # blocking / streaming user_id: str = "" bot_name: str = "AI管家" class DifyAnalysisRequest(BaseModel): # 前端 userId 可能为数字，允许 int/float 自动转 str，避免 422 model_config = ConfigDict(coerce_numbers_to_str=True) report_id: Optional[int] = None user_id: Optional[str] = None focus: Optional[str] = None messages: list[DifyMessage] = [] inputs: dict = {} class HealthCoachGenerateRequest(BaseModel): """Java 后端健康方案生成专用请求体""" family_id: Optional[int] = None member_ids: Optional[str] = None dimensions: Optional[str] = None goal: str = "" class DifyChoice(BaseModel): index: int message: dict finish_reason: str = "stop" class DifyUsage(BaseModel): prompt_tokens: int = 0 completion_tokens: int = 0 total_tokens: int = 0 class DifyResponse(BaseModel): id: str object: str = "chat.completion" created: int model: str = "langgraph-cfc" choices: list[DifyChoice] usage: DifyUsage = DifyUsage() metadata: dict = {} # ---- Helpers ---- def _now_ts() -> int: return int(time.time()) def _extract_query(req: DifyChatRequest) -> str: if req.query: return req.query # fallback: take last user message for msg in reversed(req.messages): if msg.role == "user": return msg.content return "" def _to_langgraph_context(req: DifyChatRequest) -> dict: ctx = {} if isinstance(req.inputs, dict): ctx["child_id"] = req.inputs.get("child_id") ctx["report_id"] = req.inputs.get("report_id") ctx["family_id"] = req.inputs.get("family_id") return ctx # ---- Dify-compatible endpoints ---- @router.post("/chat/completion", response_model=DifyResponse) async def chat_completion(req: DifyChatRequest): query = _extract_query(req) if not query: raise HTTPException(status_code=400, detail="query 为空") graph = create_chat_graph() initial_state = { "query": query, "user_id": int(req.user_id) if str(req.user_id).isdigit() else 0, "conversation_id": req.conversation_id or None, "intent": None, "context": _to_langgraph_context(req), "messages": None, "answer": None, "tasks": [], "sources": [], } config = { "configurable": {"thread_id": req.conversation_id or str(req.user_id or "default")}, } result = await graph.ainvoke(initial_state, config) answer = result.get("answer") or "" return DifyResponse( id=f"chatcmpl-{uuid.uuid4().hex[:24]}", created=_now_ts(), choices=[ DifyChoice( index=0, message={"role": "assistant", "content": answer}, finish_reason="stop", ) ], usage=DifyUsage( prompt_tokens=len(query.split()), completion_tokens=len(answer.split()), total_tokens=len(query.split()) + len(answer.split()), ), ) @router.post("/analysis/run", response_model=DifyResponse) async def analysis_run(req: DifyAnalysisRequest): graph = create_analysis_graph() report_id = req.report_id if report_id is None and isinstance(req.inputs, dict): report_id = req.inputs.get("report_id") initial_state = { "report_id": report_id, "user_id": int(req.user_id) if str(req.user_id).isdigit() else 0, "focus": req.focus, "report_data": None, "survey_data": None, "dimension_scores": None, "analysis": None, "recommendations": [], } result = await graph.ainvoke(initial_state, {}) analysis = result.get("analysis") or "" recommendations = result.get("recommendations") or [] content = analysis if recommendations: content += "\n\n建议:\n" + "\n".join(f"- {r}" for r in recommendations) return DifyResponse( id=f"analysis-{uuid.uuid4().hex[:24]}", created=_now_ts(), choices=[ DifyChoice( index=0, message={"role": "assistant", "content": content}, finish_reason="stop", ) ], usage=DifyUsage( prompt_tokens=len((req.focus or "").split()), completion_tokens=len(content.split()), total_tokens=len((req.focus or "").split()) + len(content.split()), ), ) @router.post("/health/coach", response_model=DifyResponse) async def health_coach(req: DifyChatRequest): query = _extract_query(req) if not query: raise HTTPException(status_code=400, detail="query 为空") graph = create_health_coach_graph() initial_state = { "query": query, "user_id": int(req.user_id) if str(req.user_id).isdigit() else 0, "child_id": int(req.inputs.get("child_id")) if req.inputs.get("child_id") else None, "conversation_id": req.conversation_id or None, "context": _to_langgraph_context(req), "answer": None, "sources": [], "memory_messages": None, } config = { "configurable": {"thread_id": req.conversation_id or f"health_{req.user_id}"}, } result = await graph.ainvoke(initial_state, config) answer = result.get("answer") or "" sources = result.get("sources") or [] metadata = {} if sources: metadata["sources"] = [ {"title": s.get("title", ""), "type": s.get("type", "knowledge")} for s in sources ] return DifyResponse( id=f"health-{uuid.uuid4().hex[:24]}", created=_now_ts(), model="langgraph-health-coach", choices=[ DifyChoice( index=0, message={"role": "assistant", "content": answer}, finish_reason="stop", ) ], usage=DifyUsage( prompt_tokens=len(query.split()), completion_tokens=len(answer.split()), total_tokens=len(query.split()) + len(answer.split()), ), metadata=metadata, ) @router.post("/health/butler", response_model=DifyResponse) async def health_butler(req: DifyChatRequest): """AI 健康管家 — 基于健康知识库检索 + 个性化建议 + 任务生成""" query = _extract_query(req) if not query: raise HTTPException(status_code=400, detail="query 为空") graph = create_health_butler_graph() # 从 inputs 中提取健康管家需要的上下文 inputs = req.inputs or {} family_id = inputs.get("family_id") child_id = inputs.get("child_id") report_id = inputs.get("report_id") focus = inputs.get("focus") # conversation_id 用于 checkpointer thread thread_id = req.conversation_id or f"butler_{req.user_id or 'anon'}" initial_state = { "query": query, "user_id": int(req.user_id) if str(req.user_id).isdigit() else 0, "conversation_id": thread_id, "family_id": int(family_id) if family_id else None, "child_id": int(child_id) if child_id else None, "report_id": int(report_id) if report_id else None, "focus": focus, "kb_context": None, "knowledge_results": [], "answer": None, "tasks": [], "sources": [], "messages": None, } config = { "configurable": {"thread_id": thread_id}, } result = await graph.ainvoke(initial_state, config) answer = result.get("answer") or "" # 从回答中提取 TASK 标记 tasks = result.get("tasks") or [] sources = result.get("sources") or [] metadata = {"tasks": tasks, "sources": sources} return DifyResponse( id=f"butler-{uuid.uuid4().hex[:24]}", created=_now_ts(), model="langgraph-health-butler", choices=[ DifyChoice( index=0, message={"role": "assistant", "content": answer}, finish_reason="stop", ) ], usage=DifyUsage( prompt_tokens=len(query.split()), completion_tokens=len(answer.split()), total_tokens=len(query.split()) + len(answer.split()), ), metadata=metadata, ) def _parse_normal_range(ref_range: str) -> tuple[float | None, float | None]: """解析正常范围字符串，返回 (下限, 上限)，如 '30-100' → (30, 100), '<5' → (None, 5)""" if not ref_range: return None, None ref_range = ref_range.strip() m = re.match(r'([<>]=?)\s*([\d.]+)', ref_range) if m: op, val = m.group(1), float(m.group(2)) if op.startswith('>'): return (val, None) else: return (None, val) m = re.match(r'([\d.]+)\s*[-~]\s*([\d.]+)', ref_range) if m: return (float(m.group(1)), float(m.group(2))) return None, None def _is_abnormal(status: str, value: float | None, low: float | None, high: float | None) -> bool: """判断指标是否异常：优先用 status 字段，否则用数值与范围比较""" if status and status not in ("正常", "正常范围", "未检出", ""): return True if value is not None and low is not None and high is not None: return value < low or value > high return False def _try_parse_value(raw: str) -> float | None: if not raw: return None raw = raw.strip().replace(",", "").replace(" ", "") try: return float(raw) except ValueError: return None async def _search_knowledge(retriever: RagRetriever, query: str, k: int = 3) -> list[dict]: """从知识库检索相关内容""" try: return await retriever.retrieve(query, k=k) except Exception as e: logger.warning("知识库检索失败: %s", e) return [] @router.post("/health/coach/generate") async def health_coach_generate(req: HealthCoachGenerateRequest): """健康方案生成 — 选人→拉指标→查知识库→LLM""" goal = req.goal or "改善健康状况" member_ids_str = req.member_ids or "" dimensions = req.dimensions or "" member_ids = [m.strip() for m in member_ids_str.split(",") if m.strip()] java = JavaClient() retriever = RagRetriever(collection_name="cfc_knowledge") llm = ChatOpenAI( model=settings.llm_model, api_key=settings.llm_api_key, base_url=settings.llm_base_url, temperature=0.3, ) # ====== 1. 获取家庭成员信息 ====== members_info = [] if member_ids: # 从 context 获取家庭信息 for uid_str in member_ids: uid = int(uid_str) ctx = await java.get_family_context(uid, "child_info") children = ctx.get("children", []) if isinstance(ctx, dict) else [] for child in children: child_id = str(child.get("用户ID", "")) if child_id in member_ids: members_info.append({ "id": child_id, "name": child.get("姓名", f"成员{child_id}"), "age": child.get("年龄", "未知"), "energy": child.get("能量", 0), }) break # 如果 context 没找到，用基本信息兜底 if not any(m["id"] == uid_str for m in members_info): members_info.append({"id": uid_str, "name": f"成员{uid_str}", "age": "未知", "energy": 0}) else: members_info.append({"id": "0", "name": "用户", "age": "未知", "energy": 0}) # ====== 2. 获取每个成员的指标数据 ====== all_indicators = [] for member in members_info: uid = int(member["id"]) reports = await java.get_member_reports(uid) if not reports: logger.info("成员 %s 无健康报告", member["id"]) continue latest = max(reports, key=lambda r: r.get("reportDate", "")) report_id = latest.get("id") member["latest_report_id"] = report_id member["report_date"] = latest.get("reportDate", "") member["overall_score"] = latest.get("overallScore", "未知") indicators = await java.get_report_indicators(report_id) for ind in indicators: ind["_member_id"] = member["id"] ind["_member_name"] = member["name"] all_indicators.append(ind) # ====== 3. 获取指标定义+正常范围 ====== known_indicators = {} for ind in all_indicators: name = ind.get("indicatorName", "").strip() if not name or name in known_indicators: continue kb = await java.query_health_knowledge("indicator", name) if not kb: kb = await java.query_health_knowledge("bacteria", name) if not kb: kb = await java.query_health_knowledge("nutrient", name) if kb: known_indicators[name] = kb # ====== 4. 识别异常指标 ====== abnormal_list = [] normal_list = [] for ind in all_indicators: name = ind.get("indicatorName", "") raw_val = ind.get("indicatorValue", "") status = ind.get("status", "") unit = ind.get("unit", "") ref_range = ind.get("refRange", "") # 优先用知识库中的正常范围 kb = known_indicators.get(name) if kb and kb.get("normalRange"): ref_range = kb.get("normalRange", ref_range) low, high = _parse_normal_range(ref_range) value = _try_parse_value(raw_val) is_abnormal = _is_abnormal(status, value, low, high) entry = { "member": ind.get("_member_name", ""), "indicator": name, "value": raw_val, "unit": unit, "ref_range": ref_range, "status": status, "is_abnormal": is_abnormal, "description": kb.get("description", "") if kb else "", "suggestion": kb.get("suggestion", "") if kb else "", } if is_abnormal: abnormal_list.append(entry) else: normal_list.append(entry) # ====== 5. 检索知识库 ====== kb_results = [] # 5a. 异常指标检索 abnormal_queries = set() for ind in abnormal_list: abnormal_queries.add(ind["indicator"]) for q in list(abnormal_queries)[:5]: results = await _search_knowledge(retriever, f"{q} 改善建议", k=3) kb_results.extend(results) # 5b. 用户需求检索 goal_results = await _search_knowledge(retriever, goal, k=5) kb_results.extend(goal_results) # 5c. 维度检索 if dimensions: dim_results = await _search_knowledge(retriever, dimensions, k=3) kb_results.extend(dim_results) # 去重 seen_content = set() deduped_kb = [] for r in kb_results: h = r.get("content", "")[:100] if h not in seen_content: seen_content.add(h) deduped_kb.append(r) # ====== 6. 组装结构化 Prompt ====== prompt_parts = [] # 系统提示 prompt_parts.append(await get_prompt("health_plan_text") or TEXT_PLAN_SYSTEM_PROMPT) # 目标与维度 prompt_parts.append(f"\n## 用户目标\n{goal}") if dimensions: prompt_parts.append(f"\n## 重点关注维度\n{dimensions}") # 成员信息 prompt_parts.append("\n## 家庭成员") for m in members_info: scores = f"健康评分: {m.get('overall_score', '未知')}" if m.get('overall_score') else "" report = f"最近报告: {m.get('report_date', '无')}" if m.get('report_date') else "" prompt_parts.append(f"- {m['name']} (年龄: {m['age']}) {scores} {report}") # 异常指标 if abnormal_list: prompt_parts.append("\n## 异常指标") for ind in abnormal_list: parts = [f"- {ind['member']} - {ind['indicator']}: {ind['value']}{ind['unit']} (参考范围: {ind['ref_range']})"] if ind['description']: parts.append(f" 说明: {ind['description']}") if ind['suggestion']: parts.append(f" 建议: {ind['suggestion']}") prompt_parts.append("\n".join(parts)) # 正常指标 if normal_list: prompt_parts.append("\n## 正常指标（参考）") normal_summary = [f"- {ind['indicator']}: {ind['value']}{ind['unit']} (正常)" for ind in normal_list[:10]] prompt_parts.extend(normal_summary) # 知识库参考 if deduped_kb: prompt_parts.append("\n## 知识库参考（可引用）") for r in deduped_kb[:8]: title = r.get("metadata", {}).get("title", "") content = r.get("content", "")[:300] prompt_parts.append(f"---\n{title}\n{content}") full_prompt = "\n".join(prompt_parts) # ====== 7. 调用 LLM ====== request_template = await get_prompt("health_plan_text_request") or "请基于以上数据，生成一份针对{goal}的健康改善方案。" try: request_text = request_template.format(goal=goal) except (KeyError, IndexError, ValueError): logger.warning("Java 配置的 health_plan_text_request 模板缺少占位符，回退本地模板") request_text = "请基于以上数据，生成一份针对{goal}的健康改善方案。".format(goal=goal) messages = [ SystemMessage(content=full_prompt), HumanMessage(content=request_text), ] response = await llm.ainvoke(messages) answer = response.content return answer # ===== 健康方案生成（结构化 JSON）===== class HealthPlanRequest(BaseModel): member_ids: Optional[str] = None dimensions: Optional[str] = None goal: str = "" family_id: Optional[int] = None class HealthPlanRegenerateRequest(BaseModel): section: str # nutrition | diet | exercise feedback: str = "" existing_section_content: str = "" member_ids: Optional[str] = None dimensions: Optional[str] = None goal: str = "" family_id: Optional[int] = None TEXT_PLAN_SYSTEM_PROMPT = """你是一个专业的家庭健康方案生成器。请根据用户提供的健康数据，生成一份结构化的健康改善方案。 输出格式要求： ## 方案概述 [简要说明方案的总体目标和适用对象] ## 成员健康概况 [每个成员的关键指标摘要] ## 需要关注的异常指标 [列出异常指标及对应的知识库建议] ## 改善方案 ### 1. 饮食调整 [具体、可执行的饮食建议] ### 2. 生活习惯 [具体、可执行的生活习惯建议] ### 3. 补充建议 [如需补充营养素或益生菌，给出具体建议] ### 4. 跟踪建议 [建议定期复查的指标和频率] ## 注意事项 [禁忌、提醒等] 请基于实际数据给出建议，不要编造科学依据。引用知识库内容时标注来源。""" PLAN_SYSTEM_PROMPT = """你是一个专业的家庭健康方案规划师。根据用户提供的健康数据和目标，生成结构化的健康改善方案。 ## 输出格式（必须输出合法 JSON，不要有其他内容） { "overview": "总体概述（100字以内，说明方案目标和核心策略）", "sections": [ { "key": "nutrition", "title": "营养补充建议", "content": "Markdown 格式的详细内容", "items": [ {"name": "产品名", "dosage": "用量", "timing": "服用时间", "reason": "推荐理由"} ], "tasks": [ { "action_type": "buy", "title": "购买维生素D3补充剂", "dimension": "wealth", "frequency": "once", "notes": "每日一粒，随餐服用" } ] }, { "key": "diet", "title": "饮食建议", "content": "Markdown 格式的餐饮建议", "items": [{"meal": "餐型", "food": "食物建议", "notes": "注意事项"}], "tasks": [ { "action_type": "diet", "title": "早餐增加高蛋白与膳食纤维", "dimension": "body", "frequency": "daily", "notes": "" } ] }, { "key": "exercise", "title": "运动计划", "content": "Markdown 格式的运动建议", "items": [{"type": "运动类型", "duration": "时长", "frequency": "频率", "notes": "注意事项"}], "tasks": [ { "action_type": "exercise", "title": "每周3次有氧运动，每次30分钟", "dimension": "body", "frequency": "daily", "notes": "" } ] } ], "abnormal_indicators": [ {"member": "姓名", "indicator": "指标名", "value": "值", "unit": "单位", "suggestion": "建议"} ] } ## tasks 字段约定 - 每个 section 的 `tasks` 是该 section 中"可执行的行动项"列表，与 `content`（人类可读 Markdown）分离。 - `action_type` 取值仅限：`buy`（购买/补充产品）、`read`（阅读）、`exercise`（运动）、`checkin`（打卡/记录）、`diet`（饮食）、`activity`（活动/社交）。 - `dimension` 取值仅限五维：`body`/`mind`/`wisdom`/`action`/`wealth`。 - `frequency`：`once`=一次性任务；`daily`=每日重复任务。 - `title` 是最终写入任务系统的标题，必须是**具体可执行的动作**，不要写纯原理/机制描述。 - 若某 section 没有可执行的行动项，`tasks` 输出空数组 `[]`。 ## 原则 1. 基于实际数据给出建议，不编造 2. 引用知识库内容时标注来源 3. 建议要具体可执行，避免空泛 4. 营养补充部分要具体到产品类型和用量 5. 严重健康问题建议咨询医生 ## 画像数据使用指南 如果提供了用户的画像数据（五维评分、身体指标、心理指标等），请结合这些真实数据给出更有针对性的建议。特别关注异常指标（如睡眠不足、压力偏高、运动频率低等），在方案中明确说明这些指标的现状和改善方向。 """ REGENERATE_SECTION_SYSTEM_PROMPT = """你是一个家庭健康方案规划师。根据用户反馈重新生成指定部分内容。 ## 输出格式（必须输出合法 JSON，不要有其他内容） { "content": "重新生成的 Markdown 内容", "tasks": [ { "action_type": "buy|read|exercise|checkin|diet|activity", "title": "可执行任务标题", "dimension": "body|mind|wisdom|action|wealth", "frequency": "once|daily", "notes": "补充说明" } ] } ## tasks 约定 - action_type 取值：buy/read/exercise/checkin/diet/activity - dimension 取值：body/mind/wisdom/action/wealth - frequency：once=一次性；daily=每日重复 - 无行动项时 tasks 输出 [] ## 原则 - 保持与原格式一致 - 结合用户反馈进行修改 - 建议要具体可执行""" async def _collect_plan_data(java: JavaClient, retriever: RagRetriever, member_ids_str: str, goal: str, dimensions: str): """统一数据收集逻辑""" member_ids = [m.strip() for m in member_ids_str.split(",") if m.strip()] # 1. 家庭成员信息 members_info = [] for uid_str in member_ids: ctx = await java.get_family_context(int(uid_str), "child_info") children = ctx.get("children", []) if isinstance(ctx, dict) else [] for child in children: cid = str(child.get("用户ID", "")) if cid == uid_str: members_info.append({ "id": cid, "name": child.get("姓名", f"成员{cid}"), "age": child.get("年龄", "未知"), }) break if not any(m["id"] == uid_str for m in members_info): members_info.append({"id": uid_str, "name": f"成员{uid_str}", "age": "未知"}) # 2. 健康指标 all_indicators = [] abnormal_list = [] for member in members_info: reports = await java.get_member_reports(int(member["id"])) if not reports: continue latest = max(reports, key=lambda r: r.get("reportDate", "")) indicators = await java.get_report_indicators(latest.get("id")) for ind in indicators: ind["_member_name"] = member["name"] all_indicators.append(ind) # 3. 知识库 kb_results = [] queries = set() for ind in all_indicators: queries.add(ind.get("indicatorName", "")) queries.add(goal) if dimensions: queries.add(dimensions) for q in list(queries)[:8]: if q: results = await retriever.retrieve(q, k=3) kb_results.extend(results) return members_info, all_indicators, kb_results # 从 LLM 原始输出中提取 tasks 的正则兜底：在 content 文本里找形如 # "1. 动作（动词/名词）..." 的行。以可识别的动作词开头视为潜在任务。 _TASK_FALLBACK_RE = re.compile(r"^\s*(?:\d+[\.、)]|\-\s*)\s*" r"(?:(?:购买|购置|阅读|看|运动|锻炼|跑步|散步|打卡|记录|饮食|吃|少|多|活动|参加|亲子).*)$") def _parse_plan_response(answer: str) -> dict: """解析 LLM 原始输出为 HealthPlanResponse；非法 JSON 或校验失败时用正则兜底提取 tasks。 返回 dict：{"success": bool, "data": {...}, "error": str|None} """ start = answer.find("{") end = answer.rfind("}") + 1 if start >= 0 and end > start: try: parsed = json.loads(answer[start:end]) resp = HealthPlanResponse.model_validate(parsed) return {"success": True, "data": resp.model_dump(), "error": None} except Exception as e: # 校验失败：尝试正则兜底 fallback = _fallback_extract_tasks(parsed if isinstance(parsed, dict) else {}) if fallback is not None: return {"success": True, "data": fallback, "error": str(e)} return {"success": False, "data": {"raw": answer, "overview": answer[:200]}, "error": str(e)} return {"success": False, "data": {"raw": answer, "overview": answer[:200]}, "error": "no JSON found"} def _fallback_extract_tasks(parsed: dict) -> Optional[dict]: """当 LLM 输出缺 tasks 或校验失败时，从各 section.content 用正则提取任务并回填。 任一 section 无 tasks 才触发；全部已含 tasks 则返回 None（表示无需兜底）。""" if not isinstance(parsed, dict): return None sections = parsed.get("sections") if not isinstance(sections, list) or not sections: return None changed = False for sec in sections: if not isinstance(sec, dict): continue tasks = sec.get("tasks") if isinstance(tasks, list) and tasks: # 检查 tasks 是否全部有效；如有无效项则视为缺失，触发 fallback try: for t in tasks: PlanTask.model_validate(t) continue # 全部有效，跳过 except Exception: pass # 有无效项，继续执行 fallback content = sec.get("content", "") extracted = [] for raw in content.split("\n"): line = raw.strip() if not line: continue m = _TASK_FALLBACK_RE.match(line) if not m: continue # 去掉行首编号/项目符号 title = re.sub(r"^\s*(?:\d+[\.、)]|\-\s*)\s*", "", line).strip() if not title: continue action = _classify_action(title) if action is None: continue extracted.append({ "action_type": action["action_type"], "title": title, "dimension": action["dimension"], "frequency": action["frequency"], "notes": "", }) if extracted: sec["tasks"] = extracted changed = True if changed: return parsed return None def _classify_action(title: str) -> Optional[dict]: """按动作词分类，映射到 action_type + 五维维度 + 频率（与 Java classifyTaskLine 对齐）。""" if re.search(r"购买|购置|采购|下单|买入|囤|选购", title): return {"action_type": "buy", "dimension": "wealth", "frequency": "once"} if re.search(r"阅读|看|读书", title): return {"action_type": "read", "dimension": "wisdom", "frequency": "daily"} if re.search(r"运动|锻炼|跑步|散步|健身|瑜伽|拉伸", title): return {"action_type": "exercise", "dimension": "body", "frequency": "daily"} if re.search(r"打卡|记录|复盘|记", title): return {"action_type": "checkin", "dimension": "mind", "frequency": "daily"} if re.search(r"饮食|吃|少|多|餐|营养|水", title): return {"action_type": "diet", "dimension": "body", "frequency": "daily"} if re.search(r"活动|参加|亲子|社交|户外|游戏", title): return {"action_type": "activity", "dimension": "action", "frequency": "daily"} return None @router.post("/health/plan/generate", response_model=dict) async def health_plan_generate(req: HealthPlanRequest): """健康方案生成 — 返回结构化 JSON（总览+营养+饮食+运动）""" goal = req.goal or "改善健康状况" java = JavaClient() retriever = RagRetriever(collection_name="cfc_knowledge") llm = ChatOpenAI( model=settings.llm_model, api_key=settings.llm_api_key, base_url=settings.llm_base_url, temperature=0.3, ) members_info, all_indicators, kb_results = await _collect_plan_data(java, retriever, req.member_ids or "", goal, req.dimensions or "") # 获取每个成员的画像数据 for member in members_info: try: profile = await java.get_member_profile(int(member["id"])) member["profile"] = profile except Exception: member["profile"] = {} # 构建 prompt parts = [await get_prompt("health_plan") or PLAN_SYSTEM_PROMPT] parts.append(f"\n## 用户目标\n{goal}") if req.dimensions: parts.append(f"\n## 重点关注维度\n{req.dimensions}") parts.append("\n## 家庭成员") for m in members_info: profile = m.get("profile", {}) dims = profile.get("dimension_scores", {}) body = profile.get("body_metrics", {}) mind = profile.get("mind_metrics", {}) parts.append(f"- {m['name']} (年龄: {m['age']})") if dims: parts.append(f" 五维评分: 身{dims.get('body','?')} 智{dims.get('wisdom','?')} 心{dims.get('mind','?')} 行{dims.get('action','?')} 富{dims.get('wealth','?')}") if body.get('sleep_dur_avg'): parts.append(f" 平均睡眠: {body['sleep_dur_avg']}小时/天") if mind.get('stress_avg'): parts.append(f" 平均压力: {mind['stress_avg']}/10") if body.get('exercise_count_week'): parts.append(f" 周运动: {body['exercise_count_week']}次") if all_indicators: parts.append("\n## 健康指标摘要") for ind in all_indicators[:15]: status = ind.get("status", "") if status in ("abnormal", "high", "low", "偏高", "偏低"): parts.append(f"- 【异常】{ind.get('_member_name','')} - {ind.get('indicatorName','')}: {ind.get('indicatorValue','')} {ind.get('unit','')} (状态: {status})") else: parts.append(f"- {ind.get('_member_name','')} - {ind.get('indicatorName','')}: {ind.get('indicatorValue','')} {ind.get('unit','')}") if kb_results: parts.append("\n## 知识库参考") for r in kb_results[:6]: title = r.get("metadata", {}).get("title", "") content = r.get("content", "")[:200] parts.append(f"---\n{title}\n{content}") full_prompt = "\n".join(parts) messages = [SystemMessage(content=full_prompt)] try: response = await llm.ainvoke(messages) answer = response.content result = _parse_plan_response(answer) if result["success"]: return {"success": True, "data": result["data"], "parse_error": result["error"]} return {"success": True, "data": result["data"], "parse_error": result["error"]} except Exception as e: logger.error("方案生成失败: %s", e) return {"success": False, "error": str(e)} @router.post("/health/plan/regenerate-section", response_model=dict) async def health_plan_regenerate(req: HealthPlanRegenerateRequest): """重新生成方案的某一个 section""" java = JavaClient() llm = ChatOpenAI( model=settings.llm_model, api_key=settings.llm_api_key, base_url=settings.llm_base_url, temperature=0.3, ) members_info, all_indicators, kb_results = await _collect_plan_data(java, None, req.member_ids or "", req.goal, req.dimensions or "") # 构建上下文 ctx_parts = [f"目标: {req.goal}"] for m in members_info: ctx_parts.append(f"- {m['name']} (年龄: {m['age']})") for ind in all_indicators[:10]: if ind.get("status") in ("abnormal", "high", "low", "偏高", "偏低"): ctx_parts.append(f"- 【异常】{ind.get('_member_name','')} - {ind.get('indicatorName','')}: {ind.get('indicatorValue','')}") prompt = await get_prompt("health_plan_regenerate") or REGENERATE_SECTION_SYSTEM_PROMPT prompt += f"\n\n## 当前 {req.section} 内容\n{req.existing_section_content[:500]}" prompt += f"\n\n## 用户反馈\n{req.feedback}" prompt += f"\n\n## 相关背景\n" + "\n".join(ctx_parts[:10]) try: response = await llm.ainvoke([SystemMessage(content=prompt)]) answer = response.content start = answer.find("{") end = answer.rfind("}") + 1 content = answer tasks = [] if start >= 0 and end > start: try: parsed = json.loads(answer[start:end]) content = parsed.get("content") or answer raw_tasks = parsed.get("tasks") or [] # 用 PlanTask 校验，非法条目丢弃 for t in raw_tasks: try: pt = PlanTask.model_validate(t) tasks.append(pt.model_dump()) except Exception: continue except Exception as e: logger.warning("解析重生成 section JSON 失败: %s", e) return {"success": True, "content": content, "tasks": tasks} except Exception as e: logger.error("重新生成 section 失败: %s", e) return {"success": False, "error": str(e)} @router.post("/nutrition/send", response_model=DifyResponse) async def nutrition_send(req: DifyChatRequest): """AI 营养助手 — 基于健康报告的个性化营养建议""" query = _extract_query(req) if not query: raise HTTPException(status_code=400, detail="query 为空") graph = create_nutrition_graph() initial_state = { "query": query, "user_id": int(req.user_id) if str(req.user_id).isdigit() else 0, "child_id": int(req.inputs.get("child_id")) if req.inputs.get("child_id") else None, "conversation_id": req.conversation_id or None, "context": _to_langgraph_context(req), "answer": None, "sources": [], "tasks": [], "messages": None, } config = { "configurable": {"thread_id": req.conversation_id or f"nutrition_{req.user_id}"}, } result = await graph.ainvoke(initial_state, config) answer = result.get("answer") or "" sources = result.get("sources") or [] metadata = {} if sources: metadata["sources"] = [ {"title": s.get("name", ""), "type": s.get("type", "tool")} for s in sources ] return DifyResponse( id=f"nutrition-{uuid.uuid4().hex[:24]}", created=_now_ts(), model="langgraph-nutrition", choices=[ DifyChoice( index=0, message={"role": "assistant", "content": answer}, finish_reason="stop", ) ], usage=DifyUsage( prompt_tokens=len(query.split()), completion_tokens=len(answer.split()), total_tokens=len(query.split()) + len(answer.split()), ), metadata=metadata, )