"""生成五维自检题目图 基于用户画像（分数、历史数据、行为数据）生成 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, }