"""用户画像生成/重画 LangGraph 输入(由 Java 传入): - user_id: 用户ID - member_id: 成员ID(可空) - previous_portrait: 旧画像文本(重画时传入,首次生成为 None) - profile_snapshot: 五维画像快照 dict(含 member/dimension_scores/body_metrics/mind_metrics/problem_domains) - behavior_summary: 行为摘要 dict 输出: - portrait_text: 生成的画像文本(中文,自然语言描述) """ from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage from app.config import settings import logging logger = logging.getLogger(__name__) class PortraitGenerationState(TypedDict): user_id: int member_id: int | None previous_portrait: str | None profile_snapshot: dict | None behavior_summary: dict | None portrait_text: str | None SYSTEM_PROMPT = """你是「浠艾福」家庭成长平台的 AI 画像分析师。 任务:根据用户的行为数据、五维能量指标、健康画像,生成一段自然语言的「用户画像」文本。 画像用途:作为后续所有 AI 对话(健康教练、管家、营养师、通用聊天)的 SystemMessage 注入,帮助 AI 更懂用户,给出更个性化的建议。 生成要求: 1. 语言自然、温暖、具体,像一位了解家庭的成长顾问写的观察笔记 2. 必须包含:核心特质、行为模式、关注点、潜在需求 3. 长度:200-500 字,不要过长 4. 不要列举原始数据(如"身 7.2 智 6.5"),要转化为洞察(如"身体维度相对突出,但智力维度有提升空间") 5. 如有 previous_portrait,重画时要体现"变化"与"延续":哪些特质稳定,哪些有新变化,原因可能是什么 输出格式:纯文本,无 JSON,无标记。""" def create_portrait_generation_graph(): llm = ChatOpenAI( model=settings.llm_model, api_key=settings.llm_api_key, base_url=settings.llm_base_url, temperature=0.7, ) builder = StateGraph(PortraitGenerationState) async def generate_portrait(state: PortraitGenerationState) -> dict: previous = state.get("previous_portrait") profile = state.get("profile_snapshot") or {} behavior = state.get("behavior_summary") or {} context_parts = [] if profile: member = profile.get("member") or {} name = member.get("name", "用户") age = member.get("age") gender = member.get("gender") context_parts.append(f"用户:{name},{age}岁,{gender}") dims = profile.get("dimension_scores") or {} if dims: context_parts.append( f"五维评分:身 {dims.get('body','?')} 智 {dims.get('wisdom','?')} " f"心 {dims.get('mind','?')} 行 {dims.get('action','?')} 富 {dims.get('wealth','?')}" ) body = profile.get("body_metrics") or {} if body.get("sleep_dur_avg"): context_parts.append(f"平均睡眠:{body['sleep_dur_avg']}h/天") if body.get("exercise_count_week"): context_parts.append(f"周运动:{body['exercise_count_week']}次") mind = profile.get("mind_metrics") or {} if mind.get("stress_avg"): context_parts.append(f"平均压力:{mind['stress_avg']}/10") problems = profile.get("problem_domains") or [] if problems: context_parts.append(f"关注问题:{', '.join(problems[:5])}") if behavior: context_parts.append("行为摘要:" + str(behavior)) context_text = "\n".join(context_parts) or "暂无额外数据" if previous: human_content = f"""【重画任务】请基于以下最新数据,重新生成用户画像。 【原画像(参考)】: {previous} 【最新数据】: {context_text} 要求:保留原画像中仍有效的核心特质,更新已变化的部分,体现"变化与延续"。""" else: human_content = f"""【生成任务】请基于以下数据,生成用户画像。 【数据】: {context_text}""" messages = [ SystemMessage(content=SYSTEM_PROMPT), HumanMessage(content=human_content), ] response = await llm.ainvoke(messages) portrait_text = response.content.strip() logger.info("生成画像完成: user_id=%d, member_id=%s, len=%d", state["user_id"], state.get("member_id"), len(portrait_text)) return {"portrait_text": portrait_text} builder.add_node("generate_portrait", generate_portrait) builder.add_edge(START, "generate_portrait") builder.add_edge("generate_portrait", END) return builder.compile() _portrait_graph = None def get_portrait_graph(): global _portrait_graph if _portrait_graph is None: _portrait_graph = create_portrait_generation_graph() return _portrait_graph