cfc/cfc-langgraph/app/graphs/health_plan_graph.py

"""
健康方案生成 LangGraph - 分步骤结构化方案生成

工作流:
1. 数据收集: 家庭成员信息 + 健康指标 + 知识库检索
2. LLM 生成: 总览概述 + 营养/饮食/运动 三个 section
3. 返回结构化 JSON

每次调用 LLM 时遵循统一的数据组装方式:
  - 用户指标 (来自 Java context API)
  - 知识库参考 (RAG 检索)
  - 系统提示词 (带输出格式模板)
"""
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from app.rag.retriever import RagRetriever
from app.config import settings
from app.tools.java_client import JavaClient
from app.prompt_service import get_prompt
import logging

logger = logging.getLogger(__name__)

SECTION_KEYS = ["nutrition", "diet", "exercise"]

DEFAULT_PROMPT = """你是一个专业的家庭健康方案规划师。根据用户的健康数据和目标，生成结构化的改善方案。

## 输出格式（必须严格遵守 JSON）

```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": "注意事项"}]
    }
  ]
}
```

## 原则
1. 基于实际数据给出建议，不编造
2. 引用知识库时标注来源
3. 建议要具体可执行，避免空泛
4. 营养补充部分要具体到品牌/产品类型和用量
5. 严重健康问题建议咨询医生
"""

REGENERATE_SECTION_PROMPT = """你是一个家庭健康方案规划师。根据用户反馈重新生成指定部分的内容。

## 当前方案内容
{existing_section_content}

## 用户反馈
{feedback}

## 相关背景数据
{context_summary}

请重新生成该部分内容，保持与原格式一致。只输出新的 content 字段值（Markdown 格式），不需要输出 JSON 结构。"""


class PlanState(TypedDict):
    member_ids: str
    dimensions: str
    goal: str
    family_id: int
    members_info: dict
    indicators: list
    abnormal_indicators: list
    indicator_changes: list
    kb_results: list
    overview: str
    nutrition_section: str
    diet_section: str
    exercise_section: str
    full_response: dict
    error: str


async def collect_data(state: PlanState) -> dict:
    """Step 1: 收集用户数据 + 知识库检索"""
    java = JavaClient()
    retriever = RagRetriever(collection_name="cfc_knowledge")

    member_ids_str = state.get("member_ids", "")
    member_ids = [m.strip() for m in member_ids_str.split(",") if m.strip()]
    goal = state.get("goal", "")
    dimensions = state.get("dimensions", "") or ""

    # 1a. 获取家庭成员信息
    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("年龄", "未知"),
                    "energy": child.get("能量", 0),
                })
                break
        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})

    # 1b. 获取健康指标
    all_indicators = []
    abnormal_list = []
    indicator_changes = []  # 成员指标"当前值 vs 上次"变化 (来自 member_indicator_current)
    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", ""))
        report_id = latest.get("id")
        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)

        # 拉取指标变化 (delta/trend), 即使无报告也可能有历史指标
        cur_list = await java.get_member_indicator_current(int(member["id"]))
        for cur in cur_list:
            delta = cur.get("delta")
            trend = cur.get("trend", "")
            if delta is not None and trend in ("up", "down"):
                indicator_changes.append({
                    "member": member["name"],
                    "member_id": member["id"],
                    "indicator": cur.get("name", ""),
                    "current": cur.get("value", ""),
                    "previous": cur.get("previousValue", ""),
                    "delta": delta,
                    "trend": trend,
                })

    # 1c. 识别异常指标
    known_indicators = {}
    for ind in all_indicators:
        name = ind.get("indicatorName", "").strip()
        if not name or name in known_indicators:
            continue
        for itype in ["indicator", "bacteria", "nutrient"]:
            kb = await java.query_health_knowledge(itype, name)
            if kb:
                known_indicators[name] = kb
                break

    for ind in all_indicators:
        name = ind.get("indicatorName", "")
        status = ind.get("status", "")
        kb = known_indicators.get(name, {})
        entry = {
            "member": ind.get("_member_name", ""),
            "indicator": name,
            "value": ind.get("indicatorValue", ""),
            "unit": ind.get("unit", ""),
            "ref_range": kb.get("normalRange", ind.get("refRange", "")),
            "description": kb.get("description", ""),
            "suggestion": kb.get("suggestion", ""),
        }
        if status in ("abnormal", "high", "low", "偏高", "偏低"):
            abnormal_list.append(entry)

    # 1c-2. 把"指标变化(delta 较大)"也视为值得关注的变化点, 并入 abnormal_list
    _change_seen = set()
    for ch in indicator_changes:
        key = (ch["member"], ch["indicator"])
        if key in _change_seen:
            continue
        _change_seen.add(key)
        try:
            d = float(ch["delta"])
        except (TypeError, ValueError):
            continue
        if abs(d) >= 10:  # 显著变化阈值 (按指标量级可调)
            arrow = "上升" if d > 0 else "下降"
            abnormal_list.append({
                "member": ch["member"],
                "indicator": ch["indicator"],
                "value": str(ch.get("current", "")),
                "unit": "",
                "ref_range": "上次: %s" % ch.get("previous", ""),
                "description": "指标较上次%s %.1f (当前 %.1f, 上次 %s)" % (
                    arrow, abs(d), float(ch["current"]) if ch.get("current") is not None else -1, ch.get("previous", "")),
                "suggestion": "该指标近期%s明显, 方案中应体现针对性调整" % arrow,
                "_is_change": True,
            })

    # 1d. 知识库检索
    kb_results = []
    queries = set()
    for ind in abnormal_list:
        queries.add(ind["indicator"])
    queries.add(goal)
    if dimensions:
        queries.add(dimensions)
    for q in list(queries)[:8]:
        results = await retriever.retrieve(q, k=3)
        kb_results.extend(results)

    return {
        "members_info": members_info,
        "indicators": all_indicators,
        "abnormal_indicators": abnormal_list,
        "indicator_changes": indicator_changes,
        "kb_results": kb_results,
    }


async def build_prompt(state: PlanState) -> str:
    """组装 LLM prompt"""
    base_prompt = await get_prompt("health_plan") or DEFAULT_PROMPT
    parts = [base_prompt]

    parts.append(f"\n## 用户目标\n{state['goal']}")
    if state.get("dimensions"):
        parts.append(f"\n## 重点关注维度\n{state['dimensions']}")

    parts.append("\n## 家庭成员")
    for m in state["members_info"]:
        parts.append(f"- {m['name']} (年龄: {m['age']})")

    if state["abnormal_indicators"]:
        parts.append("\n## 异常指标")
        for ind in state["abnormal_indicators"][:8]:
            parts.append(
                f"- {ind['member']} - {ind['indicator']}: {ind['value']}{ind.get('unit','')} "
                f"(参考: {ind['ref_range']})"
            )
            if ind.get("description"):
                parts.append(f"  说明: {ind['description']}")

    if state.get("indicator_changes"):
        parts.append("\n## 指标较上次报告的变化（方案必须考虑这些趋势）")
        for ch in state["indicator_changes"][:12]:
            arrow = "上升" if ch.get("trend") == "up" else "下降"
            parts.append(
                f"- {ch[member]} - {ch[indicator]}: {ch.get(current,)} "
                f"(上次: {ch.get(previous,)}, {arrow} {abs(ch.get(delta, 0))})"
            )

    if state["kb_results"]:
        parts.append("\n## 知识库参考")
        for r in state["kb_results"][:6]:
            title = r.get("metadata", {}).get("title", "")
            content = r.get("content", "")[:200]
            parts.append(f"---\n{title}\n{content}")

    return "\n".join(parts)


async def generate_plan(state: PlanState) -> dict:
    """Step 2: 调用 LLM 生成结构化方案"""
    llm = ChatOpenAI(
        model=settings.llm_model,
        api_key=settings.llm_api_key,
        base_url=settings.llm_base_url,
        temperature=0.3,
    )
    prompt = await build_prompt(state)
    messages = [SystemMessage(content=prompt)]
    try:
        response = await llm.ainvoke(messages)
        answer = response.content
        # 解析 JSON
        import json
        try:
            # 提取 JSON 块
            start = answer.find("{")
            end = answer.rfind("}") + 1
            if start >= 0 and end > start:
                json_str = answer[start:end]
                parsed = json.loads(json_str)
                return {"full_response": parsed, "overview": parsed.get("overview", "")}
        except (json.JSONDecodeError, Exception) as e:
            logger.warning("解析方案 JSON 失败，使用原始文本: %s", e)
        return {"full_response": {"raw": answer}, "overview": answer[:200]}
    except Exception as e:
        logger.error("LLM 生成方案失败: %s", e)
        return {"error": str(e)}


async def regenerate_section(state: PlanState) -> dict:
    """重新生成指定 section"""
    section_key = state.get("section", "nutrition")
    feedback = state.get("feedback", "")
    existing_content = state.get("existing_section_content", "")
    llm = ChatOpenAI(
        model=settings.llm_model,
        api_key=settings.llm_api_key,
        base_url=settings.llm_base_url,
        temperature=0.3,
    )

    # 构建上下文摘要
    ctx_parts = []
    for m in state.get("members_info", []):
        ctx_parts.append(f"- {m['name']} (年龄: {m['age']})")
    if state.get("goal"):
        ctx_parts.append(f"目标: {state['goal']}")
    if state.get("abnormal_indicators"):
        for ind in state["abnormal_indicators"][:5]:
            ctx_parts.append(f"- {ind['member']}: {ind['indicator']}={ind['value']}")

    regenerate_template = await get_prompt("health_plan_regenerate") or REGENERATE_SECTION_PROMPT
    try:
        prompt = regenerate_template.format(
            existing_section_content=existing_content[:500],
            feedback=feedback,
            context_summary="\n".join(ctx_parts),
        )
    except (KeyError, IndexError, ValueError):
        logger.warning("Java 配置的 health_plan_regenerate 模板缺少占位符，回退本地模板")
        prompt = REGENERATE_SECTION_PROMPT.format(
            existing_section_content=existing_content[:500],
            feedback=feedback,
            context_summary="\n".join(ctx_parts),
        )
    try:
        response = await llm.ainvoke([SystemMessage(content=prompt)])
        return {"regenerated_content": response.content}
    except Exception as e:
        logger.error("重新生成方案 section 失败: %s", e)
        return {"error": str(e)}


def create_health_plan_graph():
    builder = StateGraph(PlanState)
    builder.add_node("collect_data", collect_data)
    builder.add_node("generate_plan", generate_plan)
    builder.add_node("regenerate_section", regenerate_section)
    builder.add_edge(START, "collect_data")
    builder.add_edge("collect_data", "generate_plan")
    builder.add_edge("generate_plan", END)
    # regenerate_section 从外部直接调用，不走图
    graph = builder.compile()
    return graph