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.tools.java_client import JavaClient from app.rag.retriever import RagRetriever from app.config import settings from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage 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("""你是一个专业的家庭健康方案生成器。请根据用户提供的健康数据,生成一份结构化的健康改善方案。 输出格式要求: ## 方案概述 [简要说明方案的总体目标和适用对象] ## 成员健康概况 [每个成员的关键指标摘要] ## 需要关注的异常指标 [列出异常指标及对应的知识库建议] ## 改善方案 ### 1. 饮食调整 [具体、可执行的饮食建议] ### 2. 生活习惯 [具体、可执行的生活习惯建议] ### 3. 补充建议 [如需补充营养素或益生菌,给出具体建议] ### 4. 跟踪建议 [建议定期复查的指标和频率] ## 注意事项 [禁忌、提醒等] 请基于实际数据给出建议,不要编造科学依据。引用知识库内容时标注来源。""") # 目标与维度 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 ====== messages = [ SystemMessage(content=full_prompt), HumanMessage(content=f"请基于以上数据,生成一份针对{goal}的健康改善方案。"), ] 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 PLAN_SYSTEM_PROMPT = """你是一个专业的家庭健康方案规划师。根据用户提供的健康数据和目标,生成结构化的健康改善方案。 ## 输出格式(必须输出合法 JSON,不要有其他内容) { "overview": "总体概述(100字以内,说明方案目标和核心策略)", "sections": [ { "key": "nutrition", "title": "营养补充建议", "content": "Markdown 格式的详细内容", "items": [ {"name": "产品名", "dosage": "用量", "timing": "服用时间", "reason": "推荐理由"} ] }, { "key": "diet", "title": "饮食建议", "content": "Markdown 格式的餐饮建议", "items": [{"meal": "餐型", "food": "食物建议", "notes": "注意事项"}] }, { "key": "exercise", "title": "运动计划", "content": "Markdown 格式的运动建议", "items": [{"type": "运动类型", "duration": "时长", "frequency": "频率", "notes": "注意事项"}] } ], "abnormal_indicators": [ {"member": "姓名", "indicator": "指标名", "value": "值", "unit": "单位", "suggestion": "建议"} ] } ## 原则 1. 基于实际数据给出建议,不编造 2. 引用知识库内容时标注来源 3. 建议要具体可执行,避免空泛 4. 营养补充部分要具体到产品类型和用量 5. 严重健康问题建议咨询医生 ## 画像数据使用指南 如果提供了用户的画像数据(五维评分、身体指标、心理指标等),请结合这些真实数据给出更有针对性的建议。特别关注异常指标(如睡眠不足、压力偏高、运动频率低等),在方案中明确说明这些指标的现状和改善方向。 """ REGENERATE_SECTION_SYSTEM_PROMPT = """你是一个家庭健康方案规划师。根据用户反馈重新生成指定部分内容。 ## 输出格式 只输出新的 content 字段值(Markdown 格式字符串),不要输出 JSON 结构。 ## 原则 - 保持与原格式一致 - 结合用户反馈进行修改 - 建议要具体可执行""" 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 @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 = [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 # 解析 JSON import json try: start = answer.find("{") end = answer.rfind("}") + 1 if start >= 0 and end > start: parsed = json.loads(answer[start:end]) return {"success": True, "data": parsed} except Exception as e: logger.warning("解析方案 JSON 失败: %s", e) return {"success": True, "data": {"raw": answer, "overview": answer[:200]}, "parse_error": str(e)} 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 = 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)]) return {"success": True, "content": response.content} except Exception as e: logger.error("重新生成 section 失败: %s", e) return {"success": False, "error": str(e)}