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

import json
import logging
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import SystemMessage, HumanMessage
from app.llm.client import get_llm
from app.monitoring import monitor_agent

logger = logging.getLogger(__name__)

SYSTEM_PROMPT = """你是一位家庭健康顾问，基于五维自检结果（身·智·富·行·心，每维0-9分，满分45）给出个性化建议。

要求：
1. 对每个低分维度（≤6分）给出1-2句解读和2-3个具体可执行的微行动
2. 如有历史数据，简要对比趋势（改善/下滑）
3. 语气温暖口语化，每条解读不超过80字
4. 最后给出1句家庭整体洞察（30字以内）

返回 JSON（严格格式，不要额外文字）：
{
  "advice": [
    {
      "dimension": "mind",
      "dimensionName": "心",
      "interpretation": "你的情绪能量偏低，可能最近压力较大，建议...",
      "microActions": ["今晚睡前做10分钟深呼吸", "和伴侣约定每周一次夜谈"],
      "fallbackUsed": false
    }
  ],
  "familyInsight": "建议从行动维度入手，关系顺畅了内心才能安定"
}
"""


class GraphState(TypedDict):
    scores: dict
    question_ids: list
    user_id: int
    recent_history: list
    advice: Optional[dict]
    error: Optional[str]


class SelfCheckAnalysisAgent:
    def __init__(self):
        self.llm = get_llm()

    @monitor_agent("self_check_analysis")
    async def run(self, scores: dict, question_ids: list, user_id: int, recent_history: list) -> dict:
        try:
            score_summary = "\n".join(
                f"{v.get('dimensionName', k)}（{k}）: {v.get('score', 0)}分"
                for k, v in scores.items()
            )
            history_summary = ""
            if recent_history:
                history_summary = "历史趋势：\n" + "\n".join(
                    f"- {h.get('createdAt', '')}: 总分{h.get('totalScore', 0)}分"
                    for h in recent_history[:3]
                )
            messages = [
                SystemMessage(content=SYSTEM_PROMPT),
                HumanMessage(content=f"用户ID: {user_id}\n当前自检得分:\n{score_summary}\n{history_summary}"),
            ]
            response = await self.llm.ainvoke(messages)
            text = response.content.strip()
            if "```json" in text:
                text = text.split("```json")[1].split("```")[0].strip()
            elif "```" in text:
                text = text.split("```")[1].split("```")[0].strip()
            data = json.loads(text)
            advice_list = data.get("advice", [])
            for item in advice_list:
                item.setdefault("fallbackUsed", False)
            if advice_list:
                advice_list[0]["familyInsight"] = data.get("familyInsight", "")
            return {"advice": {"advice_json": json.dumps(advice_list, ensure_ascii=False), "fallback_used": False}}
        except Exception as e:
            logger.warning("自检建议生成失败: %s", e)
            return {"advice": {"advice_json": None, "fallback_used": True}}


def build_graph():
    agent = SelfCheckAnalysisAgent()

    def parse_input(state: GraphState) -> GraphState:
        return state

    async def call_llm(state: GraphState) -> dict:
        return await agent.run(state["scores"], state["question_ids"], state["user_id"], state["recent_history"])

    def validate(state: GraphState) -> GraphState:
        adv = state.get("advice")
        if adv is None or adv.get("advice_json") is None:
            return {**state, "error": "AI 建议生成失败"}
        return state

    graph = StateGraph(GraphState)
    graph.add_node("parse", parse_input)
    graph.add_node("llm", call_llm)
    graph.add_node("validate", validate)
    graph.add_edge(START, "parse")
    graph.add_edge("parse", "llm")
    graph.add_edge("llm", "validate")
    graph.add_edge("validate", END)
    return graph.compile()


_graph = None


def get_graph():
    global _graph
    if _graph is None:
        _graph = build_graph()
    return _graph