cfc/cfc-langgraph/app/graphs/self_check_trend_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
from app.prompt_service import get_prompt

logger = logging.getLogger(__name__)

SYSTEM_PROMPT = """分析用户近3次五维自检趋势，输出：
- aiInsight: 趋势解读（2-3句，指出最大变化维度和可能原因，口语化）
- trendSummary: 各维度 delta 简写（如"身-2 智+1 富0 行-1 心+2"）
返回 JSON：{"aiInsight": "...", "trendSummary": "..."}
"""


class TrendState(TypedDict):
    history: list
    user_id: int
    insight: Optional[dict]
    error: Optional[str]


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

    @monitor_agent("self_check_trend")
    async def run(self, history: list, user_id: int) -> dict:
        try:
            if len(history) < 2:
                return {"insight": {"aiInsight": "自检次数不足，建议完成至少2次自检后查看趋势", "trendSummary": ""}, "error": None}
            history_text = "\n".join(
                f"{h.get('createdAt', '')}: 总分{h.get('totalScore', 0)}，" +
                " ".join(f"{d.get('name','')}{d.get('score',0)}分" for d in h.get('dimensions', []))
                for h in history[-3:]
            )
            messages = [
                SystemMessage(content=await get_prompt("self_check_trend") or SYSTEM_PROMPT),
                HumanMessage(content=f"用户{user_id}的自检历史：\n{history_text}"),
            ]
            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)
            return {"insight": data, "error": None}
        except Exception as e:
            logger.warning("趋势分析失败: %s", e)
            return {"insight": {"aiInsight": "", "trendSummary": ""}, "error": str(e)}


def build_trend_graph():
    agent = SelfCheckTrendAgent()

    async def call_llm(state: TrendState) -> dict:
        return await agent.run(state["history"], state["user_id"])

    graph = StateGraph(TrendState)
    graph.add_node("llm", call_llm)
    graph.add_edge(START, "llm")
    graph.add_edge("llm", END)
    return graph.compile()


_trend_graph = None


def get_trend_graph():
    global _trend_graph
    if _trend_graph is None:
        _trend_graph = build_trend_graph()
    return _trend_graph