feat(langgraph): 新增 self_check_trend_graph 趋势分析实现

cfc-langgraph/app/graphs/self_check_trend_graph.py

@@ -0,0 +1,76 @@
+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 = """分析用户近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=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