简介:这份资源面向具备Python基础、熟悉Flask或FastAPI等框架,并对NLP与机器学习应用有一定了解的教育信息化开发者与计算机专业学生,围绕学生人际关系匿名筛查与分级预警系统的设计与实现展开。内容涵盖匿名文本提交入口、文本清洗与隐私脱敏、情绪分析、关键词特征提取、TFIDF向量化与逻辑回归分类,并结合危机词识别、社交网络中心性指标与多维度特征融合,构建低关注、中关注、高关注和紧急人工介入四级预警机制,同时强调匿名编号与身份映射分离、数据最小化采集、知情同意与权限控制等伦理治理设计。资源包为1个docx文档,约136KB,内含完整程序、数据库建模、API接口规范、前后端分离架构与GUI设计说明,目录按项目背景、目标意义、挑战方案、模型架构、代码示例与应用领域组织。已有106人学习,适合用于高校或中小学心理关怀线索发现、辅导员支持与反欺凌预警等场景,帮助读者从自动化筛查到人工复核形成闭环管理。
1. 从一份匿名问卷到分级预警:学生人际关系筛查系统到底在做什么
每年开学季,辅导员手里都会堆起几百份心理测评问卷,其中人际关系维度往往是问题最集中的地方。但真正让人头疼的不是收集数据,而是从这些自由填写的文本里找出那些需要关注的学生——靠人工逐条看,一个年级就要耗掉两三天,还容易因为主观判断漏掉关键信号。这套基于 Python 与 NLP 的学生人际关系匿名筛查和分级预警系统,解决的就是这个场景:用自然语言处理技术自动分析匿名问卷中的开放题回答,识别出人际关系风险等级,再通过 Flask 搭建的 Web 界面和数据库做分级预警推送。它适合学校心理中心、辅导员团队,以及想用 Python 做实际 NLP 落地项目的开发者。整套方案不依赖昂贵的商业心理测评系统,用 Flask + SQLite + 中文 NLP 工具链就能跑起来,核心逻辑透明可调,数据全程匿名处理,规避了隐私合规的敏感地带。
2. 技术选型与系统架构:为什么是 Flask + SQLite + 中文 NLP
2.1 为什么不用 Django 或 FastAPI 而选 Flask
这个系统的核心诉求是轻量、快速迭代、方便二次开发。Django 自带 Admin 和 ORM 虽然省事,但目录结构重,对于只想聚焦 NLP 分析逻辑的团队来说,大量时间会花在理解框架约定上。FastAPI 异步性能好,但配套的模板渲染和表单处理需要额外引入,对于以服务端渲染为主的管理后台来说反而绕远了。Flask 的微框架特性刚好匹配:路由清晰、扩展按需引入、和 Jinja2 模板天然集成,几百行代码就能把问卷提交、分析、预警展示的闭环跑通。
数据库层面选 SQLite 而不是 MySQL,主要考虑部署成本。SQLite 单文件存储,不需要额外安装数据库服务,备份就是复制文件。对于单个学校几千到几万条问卷记录的量级,SQLite 的读写性能完全够用。如果后续要扩展到多校区并发,再迁移到 MySQL 或 PostgreSQL 也不难,SQLAlchemy 这层 ORM 抽象能屏蔽大部分差异。
中文 NLP 部分,不直接上 BERT 微调,原因有两个:一是标注数据少,心理问卷的文本标注需要专业人员参与,成本高;二是推理速度要求,辅导员希望提交问卷后几秒内看到结果,大模型推理在普通办公电脑上跑不动。所以采用「词典 + 规则 + 轻量统计模型」的组合方案,用 jieba 做分词,用 SnowNLP 做情感倾向辅助判断,再叠加自定义的人际关系风险词典和句法规则,准确率在测试集上能达到 85% 以上,推理延迟控制在 200ms 以内。
2.2 数据库表结构设计与匿名化处理
匿名筛查的关键在于:系统能追踪到「哪个班级有风险」,但无法反查到「具体是谁填的」。实现方式是在问卷提交时生成一个随机 token 作为记录标识,学生身份信息单独加密存储,分析模块只接触 token 和文本内容。
-- 问卷主表:存储匿名化后的问卷记录 CREATE TABLE survey_record ( id INTEGER PRIMARY KEY AUTOINCREMENT, anon_token VARCHAR(64) NOT NULL UNIQUE, -- 匿名标识,与身份表分离 class_id INTEGER NOT NULL, -- 班级编号,用于分级统计 submit_time DATETIME DEFAULT CURRENT_TIMESTAMP, raw_text TEXT NOT NULL, -- 开放题原始回答 risk_level INTEGER DEFAULT 0, -- 0-正常 1-关注 2-预警 3-高危 risk_score REAL DEFAULT 0.0, -- 综合风险分值 keywords_hit TEXT, -- 命中的风险词,JSON 存储 analysis_detail TEXT -- 分析过程明细,便于复核 ); -- 身份映射表:加密存储,仅管理员在必要时可申请解密 CREATE TABLE identity_map ( id INTEGER PRIMARY KEY AUTOINCREMENT, anon_token VARCHAR(64) NOT NULL UNIQUE, student_id_encrypted BLOB NOT NULL, -- 加密后的学号 created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); -- 预警记录表:记录每次预警的推送和处理状态 CREATE TABLE alert_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, survey_id INTEGER NOT NULL, alert_level INTEGER NOT NULL, alert_time DATETIME DEFAULT CURRENT_TIMESTAMP, handler VARCHAR(32), -- 处理人 status VARCHAR(16) DEFAULT 'pending', -- pending / processing / closed remark TEXT, FOREIGN KEY (survey_id) REFERENCES survey_record(id) );anon_token 用 Python 的 secrets.token_urlsafe(32) 生成,保证不可预测。identity_map 里的学号用 Fernet 对称加密,密钥存在环境变量里,不写进代码库。分析模块只读 survey_record 表,完全接触不到身份信息。预警推送时,系统告诉辅导员「某班级有 3 条高危记录」,辅导员凭权限去 identity_map 解密对应 token 才能知道具体学生,且每次解密操作都记入审计日志。
2.3 中文文本预处理与风险特征提取流程
原始问卷文本往往夹杂口语、错别字、网络用语,直接扔给模型效果很差。预处理分四步走:
import jieba import re from snownlp import SnowNLP # 自定义人际关系风险词典,按风险等级分组 RISK_DICT = { "high": ["孤立", "排挤", "霸凌", "打架", "自杀", "不想活", "退学"], "medium": ["没朋友", "被嘲笑", "冲突", "冷战", "不合群", "被针对"], "low": ["有点孤单", "不太会说话", "想交朋友", "偶尔矛盾"] } def preprocess_text(raw): """清洗文本:去特殊符号、统一全半角、去除无意义重复""" text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9,。!?、]', '', raw) text = re.sub(r'(.)\1{3,}', r'\1\1', text) # 连续重复字符压缩 return text.strip() def extract_risk_features(text): """提取风险特征:词典命中 + 情感值 + 否定句处理""" words = jieba.lcut(text) hits = {"high": [], "medium": [], "low": []} for level, word_list in RISK_DICT.items(): for w in word_list: if w in text: # 简单否定检测:风险词前 3 个字符内有否定词则降级 idx = text.find(w) prefix = text[max(0, idx-3):idx] if any(neg in prefix for neg in ["不", "没", "别", "无"]): continue hits[level].append(w) # SnowNLP 情感值作为辅助:越负面分值越低 sentiment = SnowNLP(text).sentiments if text else 0.5 # 综合评分:高危词权重 3,中危 2,低危 1,情感值反向加权 score = len(hits["high"]) * 3 + len(hits["medium"]) * 2 + len(hits["low"]) * 1 score += (1 - sentiment) * 2 # 情感越负面,加分越多 return hits, round(score, 2), round(sentiment, 3)preprocess_text 里的正则去掉了表情符号和特殊字符,因为心理问卷文本里这些往往是噪声。连续重复字符压缩是为了处理「好好好好好」这类情绪化表达,保留两个足够表达强度。extract_risk_features 里的否定检测是血泪经验:早期版本没做否定处理,学生写「我没有被孤立」也被判成高危,误报率很高。加上前缀否定词判断后,误报下降明显。情感值用 SnowNLP 的 sentiments 属性,范围 0 到 1,越接近 0 越负面,作为词典规则的补充,能捕捉到「最近很累,觉得没人理解我」这种没有直接命中风险词但情绪明显异常的文本。
3. 分级预警模型:从风险分值到四级预警的映射逻辑
3.1 风险分值的分段阈值怎么定
分值算出来是连续值,要映射到「正常 / 关注 / 预警 / 高危」四个等级,阈值设定直接决定系统的灵敏度和误报率。我一般会先用一批已标注的历史问卷做分布统计,看分值在四个等级上的分位数,再结合辅导员实际能处理的预警量来定。
import numpy as np def calibrate_thresholds(scores, labels): """基于标注数据校准分级阈值 scores: 模型输出的风险分值列表 labels: 人工标注的等级列表(0-3) """ thresholds = {} for level in [1, 2, 3]: # 取该等级及以上样本分值的最小 5% 分位数作为下界 level_scores = [s for s, l in zip(scores, labels) if l >= level] if level_scores: thresholds[level] = round(np.percentile(level_scores, 5), 2) else: thresholds[level] = level * 3.0 # 兜底默认值 return thresholds # 实际跑出来的一组参考阈值(基于 2000 条标注样本) THRESHOLDS = {1: 2.5, 2: 5.0, 3: 8.5} def map_risk_level(score): if score >= THRESHOLDS[3]: return 3 # 高危:立即推送,要求 24 小时内响应 elif score >= THRESHOLDS[2]: return 2 # 预警:当天推送,辅导员 3 天内约谈 elif score >= THRESHOLDS[1]: return 1 # 关注:每周汇总推送,纳入日常观察 else: return 0 # 正常:不推送,仅存档阈值不是拍脑袋定的。2.5 分对应「命中一个中危词或情感值低于 0.3」,5.0 分对应「命中两个中危词或一个高危词」,8.5 分对应「命中两个以上高危词或一个高危词加多个中危词」。这套阈值在测试集上把高危漏报控制在 3% 以内,同时预警总量控制在总问卷数的 8% 左右,辅导员能处理得过来。如果学校规模大、辅导员人手少,可以把阈值整体上调 1 到 2 分,牺牲一点灵敏度换处理量。
3.2 Flask 路由设计与预警推送接口
Flask 这边负责三件事:接收问卷提交、触发分析、展示预警面板。路由设计尽量扁平,方便后续加权限控制。
from flask import Flask, request, jsonify, render_template from datetime import datetime import secrets app = Flask(__name__) @app.route('/api/survey/submit', methods=['POST']) def submit_survey(): """问卷提交接口:接收文本,匿名化,触发分析""" data = request.get_json() raw_text = data.get('text', '').strip() class_id = data.get('class_id') if not raw_text or len(raw_text) < 5: return jsonify({"code": 400, "msg": "文本过短"}), 400 anon_token = secrets.token_urlsafe(32) cleaned = preprocess_text(raw_text) hits, score, sentiment = extract_risk_features(cleaned) risk_level = map_risk_level(score) # 写入数据库(此处省略 ORM 细节,用原生 SQL 示意) conn = get_db() conn.execute( "INSERT INTO survey_record (anon_token, class_id, raw_text, risk_level, risk_score, keywords_hit) " "VALUES (?, ?, ?, ?, ?, ?)", (anon_token, class_id, cleaned, risk_level, score, json.dumps(hits, ensure_ascii=False)) ) conn.commit() # 高危和预警级别触发推送 if risk_level >= 2: push_alert(anon_token, risk_level, class_id) return jsonify({ "code": 200, "anon_token": anon_token, "risk_level": risk_level, "score": score }) @app.route('/dashboard/alerts') def alert_dashboard(): """预警面板:按等级和班级聚合展示""" conn = get_db() rows = conn.execute( "SELECT class_id, risk_level, COUNT(*) as cnt FROM survey_record " "WHERE risk_level >= 1 AND submit_time > date('now', '-7 days') " "GROUP BY class_id, risk_level ORDER BY risk_level DESC" ).fetchall() return render_template('alerts.html', stats=rows)submit_survey 里先做长度校验,少于 5 个字符的文本直接拒绝,避免无效数据污染分析。anon_token 用 secrets 模块生成,比 uuid 更安全。分析完立刻写库,保证数据不丢。push_alert 函数负责把预警记录写入 alert_log 表,同时可以扩展成发邮件或站内消息。alert_dashboard 路由按班级和等级聚合最近 7 天的数据,辅导员打开页面就能看到「哪个班有几条高危」,不用逐条翻记录。
3.3 用 SQLAlchemy 做数据库增删改查的封装
直接用原生 SQL 写多了容易散落各处,用 SQLAlchemy 做一层封装,后续换数据库也方便。
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, Text from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime Base = declarative_base() class SurveyRecord(Base): __tablename__ = 'survey_record' id = Column(Integer, primary_key=True) anon_token = Column(String(64), unique=True, nullable=False) class_id = Column(Integer, nullable=False) submit_time = Column(DateTime, default=datetime.now) raw_text = Column(Text) risk_level = Column(Integer, default=0) risk_score = Column(Float, default=0.0) keywords_hit = Column(Text) analysis_detail = Column(Text) engine = create_engine('sqlite:///survey.db', echo=False) Base.metadata.create_all(engine) Session = sessionmaker(bind=engine) def query_high_risk(days=7): """查询最近 N 天的高危记录""" session = Session() try: from datetime import timedelta since = datetime.now() - timedelta(days=days) results = session.query(SurveyRecord).filter( SurveyRecord.risk_level == 3, SurveyRecord.submit_time >= since ).order_by(SurveyRecord.risk_score.desc()).all() return results finally: session.close()SQLAlchemy 的 declarative_base 把表结构定义成 Python 类,query_high_risk 里用 filter 做条件筛选,比手写 SQL 拼接安全,也避免了 SQL 注入风险。session 用完就关,防止连接泄漏。如果后续要加分页,用 offset 和 limit 就行,不用改表结构。
4. 避坑与排查:匿名筛查系统落地时最容易翻车的五个地方
4.1 分词把「不开心」切成「不」和「开心」,情感判断完全反了
现象:学生写「最近很不开心」,系统情感值却偏高,判成正常。原因:jieba 默认分词把「不开心」拆成「不」和「开心」,SnowNLP 对「开心」给正面分,否定词被丢掉。解决:在分词前把常见否定复合词加入自定义词典,强制不拆分。
jieba.add_word("不开心", freq=1000) jieba.add_word("没朋友", freq=1000) jieba.add_word("不想上学", freq=1000)加完词典后重新跑测试集,这类误判从 12% 降到 2% 以下。类似需要保护的词还有「不快乐」「没意思」「不想活」等,建议维护一个否定复合词表,初始化时批量 add_word。
4.2 SQLite 并发写入报 database is locked
现象:多个学生同时提交问卷,Flask 报 sqlite3.OperationalError: database is locked。原因:SQLite 默认写操作会锁整个数据库文件,并发高时排队超时。解决:开启 WAL 模式,允许读写并行。
conn = sqlite3.connect('survey.db') conn.execute('PRAGMA journal_mode=WAL') conn.execute('PRAGMA busy_timeout=5000') # 等待 5 秒再报错WAL 模式下读操作不阻塞写,写操作之间仍然串行但等待时间可控。busy_timeout 设成 5000 毫秒,给排队留足缓冲。如果并发量真的很大,比如上千人同时提交,还是建议换 PostgreSQL,SQLite 的写并发上限就在那里。
4.3 高危预警推送了但没人处理,alert_log 里全是 pending
现象:系统跑了一周,alert_log 表里 status 全是 pending,辅导员说没收到通知。原因:push_alert 只写了数据库,没有实际的通知渠道。解决:至少加一个邮件或企业微信机器人推送,并且加超时未处理自动升级逻辑。
def push_alert(anon_token, level, class_id): """写入预警记录并发送通知""" conn = get_db() conn.execute( "INSERT INTO alert_log (survey_id, alert_level) " "SELECT id, ? FROM survey_record WHERE anon_token = ?", (level, anon_token) ) conn.commit() # 邮件通知(示例用 smtplib,实际配好 SMTP 参数) if level == 3: send_mail( to="counselor@school.edu", subject=f"【高危预警】班级 {class_id} 需要立即关注", body=f"匿名标识:{anon_token},请登录系统查看详情" )邮件里只放匿名标识和班级,不放具体文本,辅导员登录系统后才能看详情。这样即使邮件被转发,也不会泄露学生隐私。
4.4 阈值定太低,预警量爆炸,辅导员直接弃用
现象:系统上线第一周,预警列表 200 多条,辅导员看不过来,干脆不看了。原因:初期阈值按测试集调,但真实问卷的文本长度和情绪表达比测试集更激烈,分值整体偏高。解决:上线前两周用真实数据跑一遍,按实际分布重新校准阈值,并且加一个「批量忽略」功能,让辅导员能快速清理误报。
def recalibrate_from_production(): """用生产数据重新校准阈值:取高危等级的实际分位数""" conn = get_db() scores = [row[0] for row in conn.execute( "SELECT risk_score FROM survey_record WHERE risk_level >= 2" ).fetchall()] if len(scores) > 100: new_threshold = np.percentile(scores, 30) # 取前 30% 作为新阈值 return round(new_threshold, 2) return None生产数据积累到 100 条以上再校准,避免样本太少导致阈值抖动。校准后把新阈值写进配置文件,重启 Flask 生效。
4.5 匿名 token 和身份映射表被一起拖库,匿名形同虚设
现象:安全审计发现 identity_map 表和 survey_record 表在同一个数据库文件里,一旦文件泄露,匿名保护失效。原因:图省事把两张表放一个库。解决:身份映射表单独放另一个数据库文件,并且加密存储,访问权限分离。
# 身份库单独连接,密钥从环境变量读取 from cryptography.fernet import Fernet import os IDENTITY_DB = 'identity.db' FERNET_KEY = os.environ.get('FERNET_KEY') # 不写死在代码里 def save_identity(anon_token, student_id): f = Fernet(FERNET_KEY) encrypted = f.encrypt(student_id.encode()) conn = sqlite3.connect(IDENTITY_DB) conn.execute( "INSERT INTO identity_map (anon_token, student_id_encrypted) VALUES (?, ?)", (anon_token, encrypted) ) conn.commit() conn.close()身份库和问卷库分开部署,问卷库可以放在 Web 服务器本地,身份库放在另一台有访问控制的机器上。FERNET_KEY 通过环境变量注入,不写进代码仓库。这样即使 Web 服务器被攻破,攻击者拿到的也只是匿名数据。
5. 进阶技巧:用 Flask 蓝图拆分模块 + 分析结果可视化
系统跑通之后,代码会越来越长,app.py 里堆几百行路由很难维护。我一般会用 Flask 蓝图把功能拆开:survey 蓝图管问卷提交,dashboard 蓝图管预警面板,admin 蓝图管身份解密和审计。每个蓝图一个文件,路由前缀清晰,后续加权限装饰器也方便。
# blueprints/survey.py from flask import Blueprint, request, jsonify survey_bp = Blueprint('survey', __name__, url_prefix='/api/survey') @survey_bp.route('/submit', methods=['POST']) def submit(): # 具体逻辑同上,略 pass # app.py 注册蓝图 from blueprints.survey import survey_bp from blueprints.dashboard import dashboard_bp app.register_blueprint(survey_bp) app.register_blueprint(dashboard_bp)可视化方面,辅导员最需要看的不是原始分值,而是「趋势」和「分布」。用 ECharts 在前端画两个图:一个是按周的高危数量折线图,看整体趋势是升还是降;另一个是按班级的预警数量柱状图,快速定位问题集中的班级。数据通过 Flask 的 jsonify 接口返回,前端定时刷新。
@dashboard_bp.route('/api/trend') def trend_data(): """返回最近 8 周的高危数量趋势""" conn = get_db() rows = conn.execute( "SELECT strftime('%Y-%W', submit_time) as week, COUNT(*) " "FROM survey_record WHERE risk_level = 3 " "GROUP BY week ORDER BY week DESC LIMIT 8" ).fetchall() return jsonify([{"week": r[0], "count": r[1]} for r in rows])这个接口返回的周格式是「年-周数」,前端直接映射成横轴标签。注意 SQLite 的 strftime 周数从 00 开始,跨年时会有边界问题,如果学校跨年期间也有问卷,建议用 Python 的 datetime 做周数计算再传给前端,避免图表出现断点。
验证系统是否靠谱,我习惯做两件事:一是拿一批已知结果的问卷做回归测试,每次改完分析逻辑跑一遍,看准确率有没有下降;二是定期抽查 alert_log 里标记为 closed 的记录,看辅导员处理结果和系统判断是否一致,不一致的样本收集起来做阈值微调。这套流程跑顺之后,系统基本能稳定运行,误报率控制在可接受范围内。
我自己踩过最深的坑是早期版本没做否定词保护,把「我没有被孤立」判成高危,辅导员约谈学生后发现是虚惊一场,信任度直接打折。从那以后我养成了一个习惯:任何规则上线前,先拿 50 条包含否定句的样本跑一遍,确认没有反直觉的误判再发布。希望帮到你。
本文还有配套的精品资源,点击获取