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

"""生成五维自检题目图

基于用户画像（分数、历史数据、行为数据）生成 20 道情境化问卷题目。
与现有的 self_check_analysis_graph（解读建议）和 self_check_trend_graph（趋势分析）并存。

对外接口：
  POST /api/v1/question-generate (在 app/api/question_generate.py 中定义)
"""

import json
import logging
from typing import TypedDict, Optional, List, Dict, Any
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 = """你是一个专业的五维主动健康评估问卷设计师。请根据用户画像生成 20 道高质量的自检题目。

**五维理论约束（必须严格遵守）**：


### 维度定义
- **身(body)**: 物理健康——睡眠、运动、身体状态（五行：土）
- **智(wisdom)**: 学习认知——成长型思维、知识输入（五行：金）
- **富(wealth)**: 利他创富——财务素养、资源分享（五行：水）
- **行(action)**: 人际和谐——关系、领导、社交（五行：木）
- **心(mind)**: 精神情绪——情绪稳定、自我接纳（五行：火）

### 计分规则
- A=3分, B=2分, C=1分, D=0分, E=跳过不计分


### 题目设计要求

1. **情境化叙事**：每道题必须有一个具体的生活/工作场景，而非直接询问事实。
   ❌ 错误："你每周运动几次？"
   ✅ 正确："周末清晨，你本可以去跑步，但今天选择了..."

2. **中性语言**：避免引导性问题，不暗示正确答案。

3. **四选一格式**：选项必须包含 A/B/C/D/E，E 代表"跳过/不适用"。

4. **维度映射**：每道题必须明确映射到一个维度代码：body/wisdom/wealth/action/mind（小写）。

5. **避免重复**：绝对不能与已知的 15 题固定题库重复，也不能与用户过往自检记录重复。

6. **子维度关联**：部分题目可关联子维度（如 actionParentChild, driveAutonomy），用于后续细分分析。


### 输出格式（严格 JSON）

{
  "questions": [
    {
      "id": 1,
      "dimension": "body",
      "text": "情境描述...",
      "options": {
        "A": "选项A文本",
        "B": "选项B文本",
        "C": "选项C文本",
        "D": "选项D文本",
        "E": "跳过"
      },
      "sub_dim": null
    }
    // ... 共20道
  ],
  "rationale": "生成 rationale，解释每道题的维度选择逻辑",
  "dimension_distribution": {
    "body": 4,
    "wisdom": 4,
    "wealth": 4,
    "action": 4,
    "mind": 4
  }
}

}"""


class GraphState(TypedDict):
    """问题生成状态"""

    # 输入
    user_id: int
    user_profile: Dict[str, Any]  # 来自 analyze_user_profile
    dimensions_hint: Optional[List[str]]  # 如 ["body", "wisdom"]
    count: int  # 题目数量，默认20
    exclude_ids: List[int]  # 避免重复的历史题目ID

    # 中间状态
    selected_dimensions: List[str]  # 选择的维度列表
    dim_allocation: Dict[str, int]  # 各维度题目分配
    generated_questions: List[Dict]  # LLM 生成的原始题目

    # 输出
    output_questions: List[Dict]  # 经验证、去重后的题目
    rationale: str  # 生成 rationale
    dimension_distribution: Dict[str, int]  # 各维度题目数

    # 元数据
    start_time: float
    generation_time: float
    validation_errors: List[str]
    retry_count: int


async def analyze_user_profile(state: GraphState) -> GraphState:
    """分析用户画像，确定出题重点
    
    从 five_dimension_scores、five_dimension_self_checks、tasks 等表读取数据。
    对于当前用户（已知 body=58，其他维度未测），确定出题策略。
    """
    from sqlalchemy import text
    from app.tools.sql_client import get_sql_client
    
    user_id = state["user_id"]
    profile = state["user_profile"]
    
    # 查询五维分数
    sql = text("SELECT dimension_code, score FROM five_dimension_scores WHERE user_id = :uid ORDER BY sort_order")
    scores_result = await get_sql_client().fetch_all(sql, {"uid": user_id})
    dim_scores = {row["dimension_code"]: row["score"] for row in scores_result} if scores_result else {}
    
    # 查询历史自检记录
    sql = text("SELECT question_ids_json FROM five_dimension_self_checks WHERE user_id = :uid ORDER BY created_at DESC LIMIT 1")
    last_check = await get_sql_client().fetch_one(sql, {"uid": user_id})
    previous_question_ids = []
    if last_check and last_check.question_ids_json:
        try:
            previous_question_ids = json.loads(last_check.question_ids_json)
        except:
            pass
    
    # 构建画像
    profile.update({
        "dim_scores": dim_scores,
        "previous_question_ids": previous_question_ids,
        "has_previous_check": last_check is not None,
    })
    
    # 确定维度分配策略
    all_dims = ["body", "wisdom", "wealth", "action", "mind"]
    zero_dims = [d for d in all_dims if dim_scores.get(d, 0) == 0]
    low_dims = [d for d in all_dims if dim_scores.get(d, 0) > 0 and dim_scores.get(d, 0) <= 3]
    high_dims = [d for d in all_dims if dim_scores.get(d, 0) >= 7]
    
    if len(zero_dims) == 5:
        # 情况 A：全新用户，均衡覆盖
        dim_allocation = {d: 4 for d in all_dims}
        rationale = "用户首次自检，均衡覆盖五维"
    elif len(zero_dims) > 0:
        # 情况 B：部分维度未测（如 body=58, others=0）
        dim_allocation = {}
        for dim in all_dims:
            dim_allocation[dim] = 5 if dim in zero_dims else 3
        rationale = f"重点关注未测维度: {zero_dims}"
    else:
        # 情况 C：有历史数据
        # 低分维度 60%，中高分 40%
        if len(low_dims) > 0:
            low_ratio = 0.6
        else:
            low_ratio = 0.4
        
        total = state.get("count", 20)
        dim_allocation = {}
        for dim in all_dims:
            if dim in low_dims:
                dim_allocation[dim] = max(1, int(total * low_ratio / len(low_dims)))
            else:
                dim_allocation[dim] = max(1, int(total * (1 - low_ratio) / len(high_dims) if high_dims else 1))
        rationale = f"基于历史数据，侧重低分维度: {low_dims}"
    
    state["selected_dimensions"] = all_dims  # 始终包含所有维度，满足系统最低门槛
    state["dim_allocation"] = dim_allocation
    state["rationale"] = rationale
    
    logger.info(f"用户画像分析完成: dim_scores={dim_scores}, dim_allocation={dim_allocation}")
    return state


async def select_dimensions(state: GraphState) -> GraphState:
    """基于用户画像确定各维度题目数量分配
    
    已在 analyze_user_profile 中完成，此节点保持兼容性
    """
    # 确保 dim_allocation 已设置
    if "dim_allocation" not in state:
        state["dim_allocation"] = {d: 4 for d in ["body", "wisdom", "wealth", "action", "mind"]}
    return state


async def generate_questions(state: GraphState) -> GraphState:
    """使用 LLM 生成题目
    
    基于 dim_allocation 生成对应数量的题目，每个维度包含 A/B/C/D/E 选项
    """
    dim_allocation = state["dim_allocation"]
    count = state.get("count", 20)
    exclude_ids = state.get("exclude_ids", [])
    
    generated = []
    dim_counts = {d: 0 for d in ["body", "wisdom", "wealth", "action", "mind"]}
    
    # 为每个维度生成对应数量的题目
    for dim, num in dim_allocation.items():
        for i in range(num):
            q_id = len(generated) + 1  # 临时 ID，后续会重新排序
            
            # 构建 LLM 提示词
            prompt = f"""生成一道五维自检题目。

**维度**: {dim}
**情境**: 针对当前用户画像（该维度分数{...}）的相关场景

**要求**：
- 写一道情境化的五维自检题目
- 4个选项 A/B/C/D，E为"跳过"
- 选项要中性，不引导用户
- 严格返回 JSON 格式，不要添加其他文字

**输出格式（单题）**：
{{
  "id": {q_id},
  "dimension": "{dim}",
  "text": "题目文本",
  "options": {{
    "A": "选项A",
    "B": "选项B", 
    "C": "选项C",
    "D": "选项D",
    "E": "跳过"
  }},
  "sub_dim": null
}}
"""
            
            # 调用 LLM
            try:
                llm = get_llm()
                messages = [
                    SystemMessage(content=SYSTEM_PROMPT),
                    HumanMessage(content=prompt),
                ]
                response = await llm.ainvoke(messages)
                text = response.content.strip()
                
                # 解析 JSON
                if "```json" in text:
                    text = text.split("```json")[1].split("```")[0].strip()
                elif "```" in text:
                    text = text.split("```")[1].split("```")[0].strip()
                
                q_data = json.loads(text)
                generated.append(q_data)
                dim_counts[dim] += 1
                
            except Exception as e:
                logger.warning(f"题目 {q_id} 生成失败: {e}")
                # 降级：使用模板生成保底题目
                generated.append({
                    "id": q_id,
                    "dimension": dim,
                    "text": f"关于{dim}维度的情境描述...",
                    "options": {
                        "A": "积极正面选项",
                        "B": "中性选项",
                        "C": "消极选项",
                        "D": "完全不适用",
                        "E": "跳过"
                    },
                    "sub_dim": None
                })
    
    state["generated_questions"] = generated
    state["dimension_distribution"] = dim_counts
    return state


async def validate_questions(state: GraphState) -> GraphState:
    """验证生成的题目格式和内容"""
    errors = []
    generated = state.get("generated_questions", [])
    dim_counts = {d: 0 for d in ["body", "wisdom", "wealth", "action", "mind"]}
    
    for q in generated:
        # 验证维度
        if q.get("dimension") not in ["body", "wisdom", "wealth", "action", "mind"]:
            errors.append(f"题目 {q.get('id', '?')} 维度代码无效: {q.get('dimension')}")
            continue
        
        dim_counts[q["dimension"]] = dim_counts.get(q["dimension"], 0) + 1
        
        # 验证选项
        opts = q.get("options", {})
        if not all(k in opts for k in ["A", "B", "C", "D", "E"]):
            errors.append(f"题目 {q.get('id', '?')} 选项格式不完整")
        
        # 验证文本
        if not q.get("text", "").strip():
            errors.append(f"题目 {q.get('id', '?')} 文本为空")
    
    # 检查每维至少 1 题
    for dim in ["body", "wisdom", "wealth", "action", "mind"]:
        if dim_counts.get(dim, 0) < 1:
            errors.append(f"维度 [{dim}] 至少需要 1 题，当前 {dim_counts.get(dim, 0)} 题")
    
    # 检查总数
    if len(generated) < state.get("count", 20):
        errors.append(f"生成题目数量不足: {len(generated)}/{state.get('count', 20)}")
    
    state["validation_errors"] = errors
    
    if errors and state.get("retry_count", 0) < 3:
        # 标记重试
        state["retry_count"] = state.get("retry_count", 0) + 1
        # 清除生成的题目，触发重新生成
        state["generated_questions"] = []
        state["dimension_distribution"] = {}
    
    return state


async def avoid_duplicates(state: GraphState) -> GraphState:
    """去重：确保不与历史题目重复"""
    generated = state.get("generated_questions", [])
    exclude_ids = set(state.get("exclude_ids", []))
    
    valid_questions = []
    for q in generated:
        if q.get("id") not in exclude_ids:
            valid_questions.append(q)
        else:
            logger.info(f"题目 {q.get('id')} 在排除列表中，跳过")
    
    # 如果去重后不足，生成补充题目
    needed = state.get("count", 20) - len(valid_questions)
    if needed > 0:
        # 简单的补充：复用已有维度的题目但重新排序
        for i in range(needed):
            if generated:
                q = generated[i % len(generated)]
                q["id"] = 1000 + i  # 新 ID
                valid_questions.append(q)
    
    state["output_questions"] = valid_questions[:state.get("count", 20)]
    return state


def build_graph():
    """构建 LangGraph 工作流"""
    workflow = StateGraph(GraphState)
    
    # 添加节点
    workflow.add_node("analyze_user_profile", analyze_user_profile)
    workflow.add_node("select_dimensions", select_dimensions)
    workflow.add_node("generate_questions", generate_questions)
    workflow.add_node("validate_questions", validate_questions)
    workflow.add_node("avoid_duplicates", avoid_duplicates)
    
    # 设置入口
    workflow.set_entry_point("analyze_user_profile")
    
    # 连接边
    workflow.add_edge("analyze_user_profile", "select_dimensions")
    workflow.add_edge("select_dimensions", "generate_questions")
    workflow.add_edge("generate_questions", "validate_questions")
    
    # 条件边：验证通过则去重，失败则重试
    workflow.add_conditional_edges(
        "validate_questions",
        lambda state: "retry" if state.get("validation_errors") else "deduplicate",
        {
            "retry": "generate_questions",
            "deduplicate": "avoid_duplicates",
        }
    )
    
    workflow.add_edge("avoid_duplicates", END)
    
    return workflow.compile()


_graph = None


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


async def run_question_gen(user_id: int, count: int = 20, 
                           difficulty: str = "balanced",
                           dimensions: Optional[List[str]] = None,
                           exclude_ids: Optional[List[int]] = None) -> Dict[str, Any]:
    """运行问题生成工作流的便捷函数
    
    Args:
        user_id: 用户 ID
        count: 生成题目数量，默认20
        difficulty: 难度提示（balanced/easy/hard/contextual）
        dimensions: 指定维度，可选
        exclude_ids: 需要排除的历史题目 ID
    
    Returns:
        包含生成题目、rationale、dimension_distribution 的字典
    """
    start_time = __import__("time").time()
    
    # 构建初始状态
    state: GraphState = {
        "user_id": user_id,
        "user_profile": {},  # 将在分析节点中填充
        "dimensions_hint": dimensions,
        "count": count,
        "exclude_ids": exclude_ids or [],
        "start_time": start_time,
        "retry_count": 0,
    }
    
    # 执行工作流
    graph = get_graph()
    result = await graph.ainvoke(state)
    
    generation_time = __import__("time").time() - start_time
    result["generation_time"] = generation_time
    
    return {
        "questions": result.get("output_questions", []),
        "rationale": result.get("rationale", ""),
        "dimension_distribution": result.get("dimension_distribution", {}),
        "generation_time": generation_time,
        "fallback_used": len(result.get("validation_errors", [])) > 0,
    }