日期: 2026-09-30
状态: 设计稿
版本: v1.0
当前五维自检问卷存在两个核心问题:
浠艾福五维主动健康理论强调:测评不仅是打分,更是引导用户关注需要提升的维度。因此需要一套能够基于用户历史数据、当前状态、系统使用行为动态生成题目的 AI 出题系统。
QUESTION_BANK)A=3/B=2/C=1/D=0/E=跳过)submitSelfCheck 接口签名┌────────────────────────────────────────────────────────────────────────┐
│ XAF 系统整体架构 │
├────────────────────────────────────────────────────────────────────────┤
│ │
│ 小程序端 (Vue 2) │
│ └── pages/five-dimension/self-check.vue │
│ └── 调用 /api/family/self-check/ai-generate-questions │
│ └── 展示 AI 生成的题目卡片 │
│ └── 提交答案到 /api/family/self-check/submit │
│ │
├────────────────────────────────────────────────────────────────────────┤
│ cfc-backend (Java Spring Boot) │
│ ├── FiveDimensionSelfCheckController │
│ │ └── POST /api/family/self-check/ai-generate-questions │
│ │ └── 调用 LangGraphService.generateQuestions() │
│ │ │
│ └── LangGraphService.java │
│ └── generateQuestions(userId, count, difficulty) │
│ └── HTTP POST to LangGraph service │
│ │
├────────────────────────────────────────────────────────────────────────┤
│ cfc-langgraph (Python FastAPI + LangGraph) │
│ └── app/graphs/question_generation_graph.py │
│ ├── analyze_user_profile() 节点1:读取用户画像 │
│ ├── select_dimensions() 节点2:策略性选择维度 │
│ ├── generate_questions() 节点3:LLM 生成题目 │
│ ├── validate_questions() 节点4:格式校验 │
│ └── avoid_duplicates() 节点5:查重与去重 │
│ │
│ └── app/routers/langgraph.py │
│ └── POST /api/langgraph/generate-questions │
│ │
└────────────────────────────────────────────────────────────────────────┘
| 模块 | 职责 | 技术栈 |
|---|---|---|
FiveDimensionSelfCheckController |
API 入口,参数校验,调用 LangGraph | Java/Spring |
LangGraphService |
HTTP 客户端,调用 LangGraph,解析响应 | Java/OkHttp |
question_generation_graph.py |
LangGraph 工作流定义 | Python/LangGraph |
langgraph.py (router) |
REST API 端点 | Python/FastAPI |
SelfCheckQuestionVO |
题目数据传输对象 | Java |
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
from datetime import datetime
@dataclass
class QuestionGenState:
"""问题生成状态"""
# 输入
user_id: str
family_member_id: Optional[str] = None
dimensions: Optional[List[str]] = None # 如 ["body", "wisdom", "wealth", "action", "mind"]
exclude_ids: Optional[List[int]] = None
count: int = 20
difficulty_hint: str = "balanced" # balanced/easy/hard/contextual
# 中间状态
user_profile: Dict[str, Any] = field(default_factory=dict)
selected_dimensions: List[str] = field(default_factory=list)
dim_allocation: Dict[str, int] = field(default_factory=dict)
generated_questions: List[Dict] = field(default_factory=list)
# 输出
output_questions: List[Dict] = field(default_factory=list)
rationale: str = ""
dimension_distribution: Dict[str, int] = field(default_factory=dict)
# 元数据
start_time: float = 0.0
generation_time: float = 0.0
validation_errors: List[str] = field(default_factory=list)
retry_count: int = 0
graph TD
A[analyze_user_profile] --> B[select_dimensions]
B --> C[generate_questions]
C --> D{validate_questions}
D -->|失败| E[retry_count < 3?]
E -->|是| C
E -->|否| F[fill_missing_dimensions]
F --> G[avoid_duplicates]
D -->|通过| G
G --> H[output]
subgraph "用户画像分析"
A --> A1[查询 five_dimension_scores]
A --> A2[查询 five_dimension_self_checks]
A --> A3[查询 tasks 完成率]
A --> A4[构建 user_profile]
end
subgraph "维度选择策略"
B --> B1[识别低分维度 ≤3]
B --> B2[识别中高分维度 ≥7]
B --> B3[计算 dim_allocation]
end
subgraph "LLM 生成"
C --> C1[构造 prompt]
C --> C2[调用 LLM]
C --> C3[解析 JSON 响应]
end
subgraph "验证与格式化"
D --> D1[检查每维至少 1 题]
D --> D2[验证选项格式 A/B/C/D/E]
D --> D3[验证维度代码合法]
F --> F1[补充缺失维度的题目]
end
analyze_user_profileasync def analyze_user_profile(state: QuestionGenState) -> QuestionGenState:
"""
分析用户画像,确定出题重点
输入: user_id
输出: user_profile 字典
"""
# 1. 查询五维分数
dim_scores = await fetch_five_dimension_scores(state.user_id)
# 2. 查询历史自检记录
last_check = await fetch_latest_self_check(state.user_id)
previous_question_ids = last_check.question_ids_json if last_check else []
# 3. 查询行为数据(任务完成率等)
task_stats = await fetch_task_completion_stats(state.user_id)
# 4. 构建画像
state.user_profile = {
"dim_scores": dim_scores or {}, # {"body": 58, "mind": 0, ...}
"last_check_date": last_check.created_at if last_check else None,
"has_previous_check": last_check is not None,
"previous_question_ids": previous_question_ids or [],
"task_completion_rate": task_stats.get("completion_rate", 0),
"pending_tasks_count": task_stats.get("pending_tasks", 0),
}
return state
关键决策逻辑:
body >= 58 且 mind/wisdom/wealth/action ≈ 0 → 标记为 "首次深度自检" 模式,优先覆盖全部维度select_dimensionsasync def select_dimensions(state: QuestionGenState) -> QuestionGenState:
"""
基于用户画像选择维度分布策略
策略:低分维度 60% 题目,中高分维度 40% 题目
"""
dim_scores = state.user_profile.get("dim_scores", {})
# 识别维度状态
all_dims = ["body", "wisdom", "wealth", "action", "mind"]
low_dims = [d for d in all_dims if dim_scores.get(d, 0) <= 3]
high_dims = [d for d in all_dims if dim_scores.get(d, 0) >= 7]
zero_dims = [d for d in all_dims if dim_scores.get(d, 0) == 0]
# 情况 A:用户从未自检(所有维度均为 0)
if len(zero_dims) == 5:
# 均衡覆盖,每维 4 题
state.dim_allocation = {d: 4 for d in all_dims}
state.rationale = "用户首次自检,均衡覆盖五维"
# 情况 B:部分维度未测(如 body=58, others=0)
elif len(zero_dims) > 0:
# 未测维度优先,每维 5 题
# 已有分数维度,每维 3 题
for dim in all_dims:
if dim_scores.get(dim, 0) == 0:
state.dim_allocation[dim] = 5
else:
state.dim_allocation[dim] = 3
state.rationale = f"重点关注未测维度: {zero_dims}"
# 情况 C:有部分历史数据
else:
# 低分维度 60%,中高分 40%
if len(low_dims) > 0:
low_ratio = 0.6
else:
low_ratio = 0.4
state.dim_allocation = calculate_dim_allocation(
state.count, low_dims, high_dims, low_ratio
)
state.rationale = f"基于历史数据,侧重低分维度: {low_dims}"
return state
generate_questionsSYSTEM_PROMPT = """
你是一个专业的五维主动健康评估问卷设计师,服务于浠艾福平台。
## 五维理论约束(必须严格遵守)
### 维度定义
- **身(body)**: 物理健康——睡眠、运动、身体状态(五行:土)
- **智(wisdom)**: 学习认知——成长型思维、知识输入(五行:金)
- **富(wealth)**: 利他创富——财务素养、资源分享(五行:水)
- **行(action)**: 人际和谐——关系、领导、社交(五行:木)
- **心(mind)**: 精神情绪——情绪稳定、自我接纳(五行:火)
### 生克关系
- 相生:智→富→行→心→身→智
- 相克:行(克)身(克)富(克)心(克)智(克)行
### 计分规则
- A=3分, B=2分, C=1分, D=0分
- E=跳过选项(不计入 answered 计数)
- 每维度满分 9 分(3 题 × 3 分)
- 等级判定:≥7 健康 / 4-6 留意 / ≤3 紧绷
## 题目设计要求
### 1. 情境化叙事
每道题必须有一个具体的生活/工作场景,而非直接询问事实。
❌ 错误示例:"你每周运动几次?"
✅ 正确示例:"周末清晨,你本可以去跑步,但今天选择了..."
### 2. 中性语言
避免引导性问题,不暗示正确答案。
❌ 错误示例:"你应该每天运动,你做到了吗?"
✅ 正确示例:"过去一周,你的运动安排是:"
### 3. 四选一格式
选项必须包含 A/B/C/D/E,E 代表"跳过/不适用"。
### 4. 维度映射
每道题必须明确映射到一个维度代码:body/wisdom/wealth/action/mind(小写)
### 5. 子维度关联(可选)
部分题目可关联子维度(如 actionParentChild, driveAutonomy),用于后续细分分析。
### 6. 避免重复
绝对不能与已知的 15 题固定题库重复,也不能与用户过往自检记录重复。
## 输出格式(严格 JSON)
json { "questions": [
{
"id": 1,
"dimension": "body",
"text": "情境描述...",
"options": {
"A": "选项A文本",
"B": "选项B文本",
"C": "选项C文本",
"D": "选项D文本",
"E": "跳过"
},
"sub_dim": null
}
], "rationale": "生成 rationale", "dimension_distribution": {
"body": 4,
"wisdom": 4,
"wealth": 4,
"action": 4,
"mind": 4
} }
"""
async def generate_questions(state: QuestionGenState) -> QuestionGenState:
"""
使用 LLM 生成题目
输入: user_profile, dim_allocation
输出: generated_questions 列表
"""
# 构造 prompt
prompt = build_generation_prompt(state)
# 调用 LLM(通过 LangGraph 的 tool 或直接 HTTP 请求)
llm_response = await call_llm(prompt)
# 解析 JSON
try:
parsed = json.loads(llm_response)
state.generated_questions = parsed.get("questions", [])
state.rationale = parsed.get("rationale", "")
state.dimension_distribution = parsed.get("dimension_distribution", {})
except json.JSONDecodeError:
# 降级处理:使用模板生成保底题目
state.generated_questions = generate_fallback_questions(state)
state.rationale = "LLM 解析失败,使用模板降级生成"
return state
validate_questionsasync def validate_questions(state: QuestionGenState) -> QuestionGenState:
"""
验证生成的题目格式和内容
验证项:
1. 每维度至少 1 题
2. 选项格式 A/B/C/D/E 存在
3. 维度代码合法(body/wisdom/wealth/action/mind)
4. 题目文本非空
5. 不重复(与历史题目 ID 对比)
"""
errors = []
dim_counts = defaultdict(int)
for q in state.generated_questions:
# 验证维度
if q.get("dimension") not in VALID_DIMENSIONS:
errors.append(f"题目 {q['id']} 维度代码无效: {q.get('dimension')}")
continue
dim_counts[q["dimension"]] += 1
# 验证选项
if not all(opt in q.get("options", {}) for opt in ["A", "B", "C", "D", "E"]):
errors.append(f"题目 {q['id']} 选项格式不完整")
# 验证文本
if not q.get("text", "").strip():
errors.append(f"题目 {q['id']} 文本为空")
# 检查每维至少 1 题
for dim in VALID_DIMENSIONS:
if dim_counts.get(dim, 0) < 1:
errors.append(f"维度 [{dim}] 至少需要 1 题,当前 {dim_counts.get(dim, 0)} 题")
state.validation_errors = errors
state.generated_questions = state.generated_questions # 暂保留,后续去重时过滤
return state
avoid_duplicatesasync def avoid_duplicates(state: QuestionGenState) -> QuestionGenState:
"""
去重处理:确保不与历史题目重复
策略:
1. 如果历史题目 ID 在排除列表中 → 替换为该维度的新题目
2. 如果题目内容高度相似(语义相似度 > 0.8)→ 重新生成
3. 最多尝试 3 次
"""
exclude_ids = set(state.user_profile.get("previous_question_ids", []))
duplicates_found = []
valid_questions = []
for q in state.generated_questions:
if q["id"] in exclude_ids:
duplicates_found.append(q["id"])
else:
valid_questions.append(q)
# 补充缺失的题目
if len(valid_questions) < state.count:
missing_count = state.count - len(valid_questions)
supplementary = await generate_supplementary_questions(
state, missing_count, exclude_ids
)
valid_questions.extend(supplementary)
state.output_questions = valid_questions[:state.count]
return state
def question_gen_workflow():
"""
定义 LangGraph 工作流
"""
workflow = StateGraph(QuestionGenState)
# 添加节点
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.validation_errors else "deduplicate",
{
"retry": "generate_questions",
"deduplicate": "avoid_duplicates",
}
)
workflow.add_edge("avoid_duplicates", END)
return workflow.compile()
新增端点: POST /api/family/self-check/ai-generate-questions
@RestController
@RequestMapping("/api/family/self-check")
public class FiveDimensionSelfCheckController {
@Resource
private LangGraphService langGraphService;
@Operation(summary = "AI生成五维自检题目(个性化)")
@PostMapping("/ai-generate-questions")
public Result<List<SelfCheckQuestionVO>> aiGenerateQuestions(
@RequestAttribute("userId") Long userId,
@RequestBody(required = false) Map<String, Object> body) {
if (userId == null) {
return Result.error("请先登录");
}
try {
// 1. 解析参数
Integer count = body != null && body.get("count") != null
? Integer.valueOf(body.get("count").toString()) : 20;
String difficulty = body != null && body.get("difficulty") != null
? body.get("difficulty").toString() : "balanced";
List<String> dimensions = body != null && body.get("dimensions") != null
? (List<String>) body.get("dimensions") : null;
// 2. 调用 LangGraph
Map<String, Object> result = langGraphService.generateQuestions(
userId, count, difficulty, dimensions
);
// 3. 转换为 VO
@SuppressWarnings("unchecked")
List<SelfCheckQuestionVO> questions = (List<SelfCheckQuestionVO>) result.get("questions");
return Result.success(questions);
} catch (IllegalArgumentException e) {
return Result.error(e.getMessage());
} catch (Exception e) {
log.error("AI 生成题目失败: {}", e.getMessage(), e);
return Result.error("生成失败,请稍后重试");
}
}
}
LangGraphService.java(新增服务类):
@Service
public class LangGraphService {
private static final Logger log = LoggerFactory.getLogger(LangGraphService.class);
@Value("${langgraph.base.url:http://127.0.0.1:9000}")
private String langGraphBaseUrl;
/**
* 调用 LangGraph 生成五维自检题目
*/
public Map<String, Object> generateQuestions(
Long userId, Integer count, String difficulty, List<String> dimensions) {
// 构造请求体
Map<String, Object> requestBody = new HashMap<>();
requestBody.put("user_id", String.valueOf(userId));
requestBody.put("count", count);
requestBody.put("difficulty", difficulty);
if (dimensions != null && !dimensions.isEmpty()) {
requestBody.put("dimensions", dimensions);
}
// 调用 LangGraph
String url = langGraphBaseUrl + "/api/langgraph/generate-questions";
String response = HttpUtils.post(url, requestBody);
// 解析响应
try {
Map<String, Object> respMap = objectMapper.readValue(response, Map.class);
return respMap;
} catch (Exception e) {
log.error("解析 LangGraph 响应失败: {}", e.getMessage());
throw new RuntimeException("AI 服务异常", e);
}
}
}
新增端点: POST /api/langgraph/generate-questions
# app/routers/langgraph.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
router = APIRouter()
class GenerateQuestionsRequest(BaseModel):
user_id: str
count: int = 20
difficulty: str = "balanced" # balanced/easy/hard/contextual
dimensions: Optional[List[str]] = None # 可选,如 ["body", "wisdom"]
exclude_ids: Optional[List[int]] = None
class GenerateQuestionsResponse(BaseModel):
questions: List[Dict[str, Any]]
total_generated: int
generation_time: float
rationale: str
dimension_distribution: Dict[str, int]
@router.post("/generate-questions")
async def generate_questions(request: GenerateQuestionsRequest) -> GenerateQuestionsResponse:
"""
AI 生成五维自检题目
请求体:
- user_id: 用户 ID
- count: 生成题目数量,默认 20
- difficulty: 题目难度/复杂度,默认 "balanced"
- dimensions: 指定维度,可选
- exclude_ids: 需要排除的题目 ID
响应体:
- questions: 题目列表
- total_generated: 实际生成数量
- generation_time: 生成耗时(秒)
- rationale: 生成 rationale
- dimension_distribution: 各维度题目数量分布
"""
try:
# 构建状态
state = QuestionGenState(
user_id=request.user_id,
count=request.count,
difficulty_hint=request.difficulty,
dimensions=request.dimensions,
exclude_ids=request.exclude_ids,
start_time=time.time()
)
# 执行工作流
result = await question_gen_workflow().ainvoke(state)
# 构造响应
response = GenerateQuestionsResponse(
questions=result["output_questions"],
total_generated=len(result["output_questions"]),
generation_time=result["generation_time"],
rationale=result["rationale"],
dimension_distribution=result["dimension_distribution"]
)
return response
except Exception as e:
log.error(f"AI 生成题目失败: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"生成失败: {str(e)}")
以下是为 body=58, 其余维度为 0 的用户设计的示例题目,体现"情境化"和"针对性":
Q1(已有:Q1,不重复,场景变化)
Q2(已有:Q2 的变体)
Q3(新增)
Q4(新增)
Q5(新增)
Q6(新增)
Q7(新增)
Q8(新增)
Q9(新增)
Q10(新增)
Q11(新增)
Q12(新增)
Q13(新增)
Q14(新增)
Q15(新增)
Q16(子维度:actionParentChild,新增)
Q17(新增)
Q18(新增)
Q19(新增)
Q20(子维度:selfConceptOverall,新增)
| 场景 | 使用的题目 | 说明 |
|---|---|---|
| 用户首次自检 | 15 题固定题库 + AI 生成 5 题 → 总计 20 题 | 满足系统"≥10 题"门槛 |
| 用户再测 | AI 完全新的 20 题(通过 exclude_ids 避免重复) |
确保新鲜感 |
| 系统默认流程 | 继续 15 题固定题库 | 向后兼容 |
| 高级用户选项 | 可选"AI 深度自检"20 题 | 提供差异化体验 |
选项格式:严格保持 A/B/C/D/E 格式,对应 scoreOf() 方法:
"A" → 3 分"B" → 2 分"C" → 1 分D/E → 0 分(E 跳过不计入 answered)维度代码:必须为 body/wisdom/wealth/action/mind(小写),对应 DIMENSION_META
计分公式:
dimScore = round(sum * 3.0 / answered) capped at 9
totalScore = sum of all dimScore (max 45)
LangGraph 服务不可用
↓
降级到 Java 本地题库生成(随机从 FULL_QUESTION_POOL 抽取 20 题)
↓
仍需满足"每维至少 1 题"的约束
实现位置:LangGraphService.generateQuestions() 中捕获异常后自动降级。
cfc-langgraph/app/graphs/ 新建 question_generation_graph.py/api/langgraph/generate-questions 端点LangGraphService.javaFiveDimensionSelfCheckController 新增 POST /api/family/self-check/ai-generate-questions/submit 兼容)| 风险 | 等级 | 应对策略 |
|---|---|---|
| AI 生成题目质量不达标 | 中 | 严格的 prompt 工程 + 自动化验证 + 人工复核首批次 |
| 与现有 15 题冲突 | 低 | exclude_ids 机制 + 内容相似度检测 |
| LangGraph 服务不可用 | 高 | 降级到 Java 本地题库生成 |
| 用户界面体验差 | 中 | 与前端协同设计 + A/B 测试 |
| 文件 | 关联点 | 说明 |
|---|---|---|
FiveDimensionSelfCheckService.java |
buildQuestionBank() |
新增的 AI 题目不替换此方法,而是作为扩展 |
FiveDimensionSelfCheckController.java |
新增 /ai-generate-questions |
保留现有所有端点不变 |
WuxingSourcingService.java |
DIMENSION_META |
AI 生成题目需引用此映射 |
langgraph-service |
新增 question_generation_graph.py |
核心工作流实现 |
SelfCheckQuestionVO.java |
题目数据传输 | 无需修改,直接复用 |
| 决策 | 结论 | 理由 |
|---|---|---|
| AI 题目是否替换 15 题固定题库? | 否 | 保持向后兼容,作为"可选深度模式" |
| 每道题必须有 E 选项吗? | 是 | 符合系统 E=skip 逻辑 |
| 子维度题目如何计分? | A=100/B=66/C=33/D=0 | 沿用现有 subScoreOf() 逻辑 |
| 降级策略是什么? | 本地题库随机抽取 | 保证服务可用性 |
| 题目 ID 如何处理? | AI 生成非连续 ID | 避免与固定题库 (1-15) 冲突 |
文档状态: 设计稿
下一步: 评审通过后开始阶段 1 实施
负责人: [待分配]