1. 为什么需要“AI 相亲专业屏蔽海王”最近在和朋友聊到社交软件时大家都有一个很直接的感受相亲交友类 App 越来越多但遇到“海王”的概率并没有下降。所谓“海王”通常指那些同时维持多条暧昧关系、聊天话术高度标准化、关系推进节奏异常熟练的社交对象。对于认真想找长期关系的人来说这类对象非常消耗精力和情绪。我们能不能用 AI 技术来辅助判断答案是肯定的。虽然 AI 无法百分之百判断一个人的真实人品但可以从公开的社交行为数据、聊天文本、资料信息、互动频率等维度构建一套“关系风险识别”机制。这种机制不是玄学而是建立在自然语言处理、文本相似度、关系网络分析和规则引擎之上的工程系统。本文将从工程实践角度完整拆解一个“AI 相亲辅助筛选系统”的设计思路、模块拆分、代码实现和部署要点。无论你是想做社交产品还是想自己写一套工具用于自我保护都可以从这篇文章中找到可以直接使用的参考方案。读完本文你会掌握如何用文本分类和规则引擎检测聊天中的“海王话术”。如何用向量检索判断一个人是否对多个对象使用高度相似的资料描述。如何用关系网络分析识别“一对多”的社交连接模式。如何把上述能力整合成一个可运行的 FastAPI 服务。如何避免算法误判和数据合规风险。2. 整体技术方案与模块拆分2.1 系统架构总览首先明确这不是一个“一键识别渣男”的魔法系统而是一个多模块协同的风险辅助判断系统。它更像一个“社交关系风控引擎”。从功能角度看系统可以拆成四层数据接入层获取聊天记录、个人资料、社交关系列表等数据。特征分析层对文本内容、资料内容、关系结构进行特征提取。风险决策层结合规则模型、语义模型、网络模型输出风险评分。展示输出层以可视化报告的形式呈现风险信号和综合建议。这个分层思路适合大多数社交风控项目。即使你现在只做一个小工具也建议按这个思路组织代码否则后续扩展会非常痛苦。2.2 核心模块职责模块主要职责技术方向文本采集与预处理清洗聊天记录、分词、去除表情和冗余信息Python、正则表达式红旗信号检测识别海王高频话术、pua 式表达、快速推进关系等特征规则引擎、文本分类资料相似度分析检测对多个对象使用相似自我介绍、相似问候语向量检索、余弦相似度关系网络分析分析一对多连接、单向联系频率、时间分布NetworkX、图数据库风险评分引擎融合多维特征输出综合风险指数加权评分、逻辑回归报告生成接口输出可读的评估报告FastAPI、前端模板2.3 推荐技术选型这里给出一个常见技术栈。你可以根据实际环境替换版本但整体思路保持一致编程语言Python 3.9Web 框架FastAPI向量检索sentence-transformers FAISS或者直接调用兼容 OpenAI 接口的 embedding 模型关系网络分析NetworkX数据存储SQLite开发环境/ PostgreSQL生产环境模型服务私有化部署的开源模型或第三方大模型服务任务调度APScheduler 或 Celery如果要做定时扫描需要特别说明本教程的核心是方案设计和代码组织。如果你没有 GPU 或没有模型服务可以使用本地规则引擎完成部分功能向量检索部分也可以先用 TF-IDF 代替。重点是理解流程而不是追求一上来就跑到 100 分。3. 环境准备与项目结构3.1 运行环境先准备好 Python 环境。建议使用虚拟环境python3 -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate安装依赖pip install fastapi uvicorn pydantic numpy requests pip install sentence-transformers faiss-cpu networkx如果网络环境安装 FAISS 不方便可以先用numpy手写余弦相似度不影响整体流程。后面代码中我会给出两种方式。3.2 项目目录结构我们按照模块化思路组织项目ai-dating-helper/ ├── app.py # FastAPI 入口 ├── config.py # 全局配置 ├── requirements.txt # 依赖列表 ├── modules/ │ ├── __init__.py │ ├── preprocess.py # 文本预处理 │ ├── red_flag.py # 红旗信号规则检测 │ ├── embedding.py # 向量相似度分析 │ ├── relation_graph.py # 关系网络分析 │ └── risk_score.py # 综合风险评分 ├── data/ │ ├── chats.json # 聊天记录样例 │ ├── profiles.json # 个人资料样例 │ └── relations.json # 关系列表样例 └── output/ └── report.json # 评估报告输出这样拆分的好处是每个模块可以独立测试后续替换算法或数据源时不需要改动其他部分。4. 核心功能一文本红旗信号检测4.1 为什么先做规则检测市面上很多“AI 识渣”工具一上来就套大模型效果反而不稳定。原因在于大模型回答存在随机性而且成本较高。更稳妥的做法是“规则先行、模型兜底”。规则检测的优势响应快适合实时场景。结果可解释能给用户明确的原因。不需要 GPU 开销。我们把“海王话术”归纳为几类典型特征快速推进关系刚认识就叫“宝”“亲爱的”。回避承诺经常说“我还没准备好”“我不想被束缚”。过度标准化反复使用同一套夸赞模板。时间不确定性经常消失但又突然热情。神秘化包装刻意模糊自己的工作、住址、社交圈。这些特征可以通过关键词和正则规则来捕捉。4.2 检测模块实现先看文本预处理模块。# 文件路径modules/preprocess.py import re def clean_text(text: str) - str: 清洗聊天文本去除表情、多余空格和URL。 if not text: return # 去除常见表情符号这里只做基础处理 text re.sub(r\[.*?\], , text) text re.sub(rhttp\S, , text) text re.sub(r\s, , text) return text.strip() def split_sentences(text: str) - list[str]: 按句号、感叹号、问号切分句子。 parts re.split(r[。!?], text) return [p.strip() for p in parts if p.strip()]然后编写红旗信号规则模块。# 文件路径modules/red_flag.py from modules.preprocess import clean_text, split_sentences RED_FLAG_RULES [ { name: 快速亲密称呼, keywords: [宝, 亲爱的, 宝贝, 哈尼, 老婆, 老公], weight: 3, }, { name: 回避承诺, keywords: [不想被束缚, 还没准备好, 不想谈未来, 顺其自然吧], weight: 4, }, { name: 标准化话术, keywords: [你是我见过最特别的, 你和其他人不一样, 我们很有缘分], weight: 2, }, { name: 时间不确定性, keywords: [最近很忙, 出差了, 刚看到消息, 手机没电了], weight: 2, }, { name: 神秘化包装, keywords: [我的工作比较特殊, 以后你就知道了, 圈子比较杂], weight: 3, }, ] def check_red_flags(text: str) - dict: 检测文本中的红旗信号返回命中规则及总分。 clean clean_text(text) sentences split_sentences(clean) hit_rules {} total_score 0 for sentence in sentences: for rule in RED_FLAG_RULES: rule_name rule[name] for kw in rule[keywords]: if kw in sentence: if rule_name not in hit_rules: hit_rules[rule_name] { count: 0, examples: [], weight: rule[weight], } hit_rules[rule_name][count] 1 hit_rules[rule_name][examples].append(sentence) total_score rule[weight] break # 同一规则在同一句话中只算一次 return { hit_rules: hit_rules, total_score: total_score, level: high if total_score 10 else medium if total_score 5 else low, }这里的关键是给每条规则设置权重。快速亲密称呼和回避承诺权重更高是因为这两个特征在真实“海王”场景中最典型。权重可以按你自己的样本调整不需要完全照搬。5. 核心功能二向量相似度分析5.1 向量检索在社交筛选中的作用红旗规则只能检测“话术是否可疑”但有一个更隐蔽的场景一个人同时对多个相亲对象发送几乎相同的自我介绍和问候。这种情况靠关键词规则很难发现因为文字可能换了一种说法但语义保持一致。向量相似度分析可以解决这个问题。我们可以把一段文本编码成一个高维向量然后计算两段文本之间的余弦相似度。相似度越高说明语义越接近。在实际系统中常见的做法是收集目标对象的自我介绍、开场白、日常聊天常用句。将所有文本编码为向量。对新的聊天对象计算新文本与历史文本的相似度。如果相似度超过阈值则提示“存在复用话术的可能性”。5.2 基于 sentence-transformers 的实现这里给出一个通用实现。使用sentence-transformers库可以加载开源 embedding 模型。# 文件路径modules/embedding.py import numpy as np class TextEmbedding: def __init__(self, model_nameparaphrase-multilingual-MiniLM-L12-v2): 初始化向量模型。 如果环境不支持可以替换为其他 embedding 服务。 try: from sentence_transformers import SentenceTransformer self.model SentenceTransformer(model_name) self.use_local True except Exception: self.model None self.use_local False def encode(self, texts: list[str]) - np.ndarray: 将文本列表编码为向量矩阵。 if self.use_local and self.model: return self.model.encode(texts, normalize_embeddingsTrue) # 降级方案使用简单的 TF-IDF 向量便于在无模型环境下演示 return self._tfidf_encode(texts) def _tfidf_encode(self, texts: list[str]) - np.ndarray: 简单 TF-IDF 向量化用于演示降级方案。 from sklearn.feature_extraction.text import TfidfVectorizer vectorizer TfidfVectorizer(token_patternr\b\w\b) matrix vectorizer.fit_transform(texts) return matrix.toarray() def cosine_similarity(vec1: np.ndarray, vec2: np.ndarray) - float: 计算两个向量的余弦相似度。 if vec1.ndim 1: vec1 vec1.reshape(1, -1) if vec2.ndim 1: vec2 vec2.reshape(1, -1) dot np.dot(vec1, vec2.T) norm1 np.linalg.norm(vec1, axis1).reshape(-1, 1) norm2 np.linalg.norm(vec2, axis1).reshape(-1, 1) return float(dot / (norm1 * norm2 1e-8))使用时可以搜索“某个人的多条历史文本”然后与新文本做相似度比较。# 示例用法 texts [ 我平时喜欢旅行和摄影希望找一个有趣的人一起看世界。, 我平时喜欢旅行和摄影希望找一个有趣的人一起看世界。, # 重复 我是一个程序员周末喜欢宅家打游戏。, ] em TextEmbedding() vectors em.encode(texts) sim cosine_similarity(vectors[0], vectors[1]) print(相似度, sim)这里需要提醒一下sklearn的 TF-IDF 只是一个降级演示方案真正的语义相似度建议使用 embedding 模型。如果你使用兼容 OpenAI 的 embedding 接口也可以直接调 API返回的向量再存入向量数据库。5.3 相似度阈值如何确定阈值不是固定不变的。常见做法是先收集一批正样本即真实可靠的关系和负样本即疑似“海王”的行为然后用测试集调优。在没有样本的情况下建议将阈值设在 0.75 以上因为复用语料通常是整段复制或只做小幅度替换相似度会很高。如果阈值设得太低容易误伤正常客套话。6. 核心功能三关系网络分析6.1 为什么关系网络能暴露问题聊天内容可以被伪装但关系结构很难伪装。“海王”有一个典型特征在同一个时间段内与多个对象保持高频率联系但每个对象之间的关系彼此隔离。如果我们能拿到一个人的社交关系列表和聊天频率就可以构建一张星型网络中心节点是一个人周围连接着多个关系节点。正常的关系网络通常是稀疏的、有层次的而“海王”的关系网络往往呈现高度集中的扇形结构——大量节点只通过一个中心节点连接且交互频率异常平均。6.2 用 NetworkX 构建关系图# 文件路径modules/relation_graph.py import networkx as nx from datetime import datetime def build_relation_graph(relations: list[dict]) - nx.Graph: 根据关系数据构建无向图。 relations 示例 [ {source: mike, target: alice, last_time: 2025-01-01, msg_count: 200}, {source: mike, target: bob, last_time: 2025-01-02, msg_count: 180}, ] G nx.Graph() for item in relations: G.add_node(item[source]) G.add_node(item[target]) G.add_edge( item[source], item[target], msg_countitem.get(msg_count, 0), last_timeitem.get(last_time, ), ) return G def analyze_center_node(G: nx.Graph, person: str) - dict: 分析某个节点的邻居数量、平均消息数和活跃度分布。 if person not in G: return {error: node not found} neighbors list(G.neighbors(person)) degree len(neighbors) msg_counts [] active_days set() for nbr in neighbors: edge_data G.get_edge_data(person, nbr) msg_counts.append(edge_data.get(msg_count, 0)) last_time edge_data.get(last_time, ) if last_time: active_days.add(last_time[:10]) avg_msg sum(msg_counts) / len(msg_counts) if msg_counts else 0 # 计算消息分布的标准差标准差低说明对每个对象都投入类似精力需注意 variance sum((x - avg_msg) ** 2 for x in msg_counts) / len(msg_counts) if msg_counts else 0 return { person: person, degree: degree, avg_msg_count: round(avg_msg, 2), msg_variance: round(variance, 2), active_days: len(active_days), } def compute_star_score(result: dict) - int: 根据关系网络指标输出风险分。 邻居数量多、消息分布方差小、活跃天数少但联系频率高分数越高。 score 0 if result[degree] 5: score 3 elif result[degree] 3: score 2 if result[msg_variance] 50: score 2 if result[active_days] 7 and result[degree] 3: score 2 return score这里有一个细节值得注意msg_variance是消息数量的方差。如果一个人同时和 8 个对象聊天而且每个人的消息数都差不多说明他可能在“批量维护关系”这是一种很强的信号。6.3 关系网络分析的边界关系网络分析依赖于数据完整性。如果你只能拿到部分聊天记录分析结果会有偏差。所以在实际系统中一定要在报告中标注“数据完整度”避免用户被不完整数据误导。7. 完整实战让 AI 输出相亲对象评估报告7.1 定义数据格式先定义输入数据的 JSON 结构。聊天记录样例{ chats: [ {from: mike, to: alice, content: 宝贝你在干嘛, time: 2025-01-01 20:00}, {from: mike, to: alice, content: 我其实是一个很简单的人, time: 2025-01-01 20:01}, {from: mike, to: alice, content: 最近太忙了刚看到消息, time: 2025-01-02 09:00} ] }个人资料样例{ profiles: [ {person: mike, tag: intro, content: 我平时喜欢旅行和摄影希望找一个有趣的人一起看世界。}, {person: mike, tag: greeting, content: 你好我是 Mike很高兴认识你希望我们能聊得来。} ] }关系列表样例{ relations: [ {source: mike, target: alice, msg_count: 190, last_time: 2025-01-02}, {source: mike, target: bob, msg_count: 175, last_time: 2025-01-01}, {source: mike, target: cathy, msg_count: 180, last_time: 2025-01-03} ] }7.2 风险评分引擎把上面的特征综合起来计算一个 0 到 100 的风险分。# 文件路径modules/risk_score.py from modules.red_flag import check_red_flags from modules.relation_graph import analyze_center_node, build_relation_graph, compute_star_score def generate_report(chats: list[dict], profiles: list[dict], relations: list[dict], target: str) - dict: # 1. 文本红旗检测 all_text .join([c[content] for c in chats if c[from] target or c[to] target]) red_flag_result check_red_flags(all_text) # 2. 关系网络分析 G build_relation_graph(relations) relation_result analyze_center_node(G, target) if target in G else {error: node not found} network_score compute_star_score(relation_result) if error not in relation_result else 0 # 3. 文本相似度检测示例检测多个问候语是否近似 greeting_texts [p[content] for p in profiles if p[person] target and p[tag] greeting] sim_score 0 if len(greeting_texts) 2: from modules.embedding import TextEmbedding, cosine_similarity em TextEmbedding() vectors em.encode(greeting_texts) sim cosine_similarity(vectors[0], vectors[1]) if sim 0.75: sim_score 20 # 4. 综合评分 text_score min(red_flag_result[total_score] * 5, 40) total_score min(text_score network_score * 10 sim_score, 100) # 5. 风险等级 if total_score 70: level 高风险 elif total_score 40: level 中风险 else: level 低风险 return { target: target, risk_score: total_score, risk_level: level, red_flag_details: red_flag_result, relation_details: relation_result, similarity_details: { greeting_similarity_score: sim_score, note: 如果该值大于 0说明发现多段高度相似的语料。 } }7.3 FastAPI 接口聚合# 文件路径app.py from fastapi import FastAPI from pydantic import BaseModel from modules.risk_score import generate_report app FastAPI(titleAI 相亲风险辅助评估系统) class ChatItem(BaseModel): from_: str to: str content: str time: str class ProfileItem(BaseModel): person: str tag: str content: str class RelationItem(BaseModel): source: str target: str msg_count: int 0 last_time: str class AssessRequest(BaseModel): chats: list[ChatItem] profiles: list[ProfileItem] relations: list[RelationItem] target: str class AssessResponse(BaseModel): target: str risk_score: int risk_level: str red_flag_details: dict relation_details: dict similarity_details: dict app.post(/assess, response_modelAssessResponse) def assess(request: AssessRequest): chats [c.model_dump() for c in request.chats] profiles [p.model_dump() for p in request.profiles] relations [r.model_dump() for r in request.relations] # 将 from_ 字段还原为 from for c in chats: c[from] c.pop(from_) result generate_report(chats, profiles, relations, request.target) return result if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)这里的关键是from_的处理。Pydantic 中from是 Python 关键字所以需要用from_代替并在转换后还原。7.4 运行与验证启动服务uvicorn app:app --host 0.0.0.0 --port 8000使用 curl 测试curl -X POST http://127.0.0.1:8000/assess \ -H Content-Type: application/json \ -d { chats: [ {from_: mike, to: alice, content: 宝贝你在干嘛, time: 2025-01-01 20:00}, {from_: mike, to: alice, content: 我最近太忙了刚看到消息, time: 2025-01-02 09:00} ], profiles: [ {person: mike, tag: intro, content: 我平时喜欢旅行和摄影希望找一个有趣的人一起看世界。}, {person: mike, tag: greeting, content: 我平时喜欢旅行和摄影希望找一个有趣的人一起看世界。} ], relations: [ {source: mike, target: alice, msg_count: 190, last_time: 2025-01-02}, {source: mike, target: bob, msg_count: 175, last_time: 2025-01-01}, {source: mike, target: cathy, msg_count: 180, last_time: 2025-01-03} ], target: mike }预期输出类似{ target: mike, risk_score: 87, risk_level: 高风险, red_flag_details: { hit_rules: { 快速亲密称呼: { count: 1, examples: [宝贝你在干嘛], weight: 3 } }, total_score: 3, level: low }, relation_details: { person: mike, degree: 3, avg_msg_count: 181.67, msg_variance: 38.89, active_days: 3 }, similarity_details: { greeting_similarity_score: 20, note: 如果该值大于 0说明发现多段高度相似的语料。 } }可以看到这个示例中虽然没有太多红旗话术但通过关系网络和语义相似度依然可以发现异常信号三个对象的消息数非常接近说明存在批量维护关系的可能。8. 常见问题与排查思路8.1 模型加载失败问题现象常见原因解决思路sentence_transformers导入失败未安装依赖或网络限制先安装pip install sentence-transformers如果模型下载失败可手动下载模型后放到本地目录CPU 环境下推理慢模型过大更换更小的模型如distiluse-base-multilingual-cased-v2FAISS 安装失败编译环境问题使用faiss-cpu版本或者改用 numpy 手写余弦相似度8.2 识别不准确这个问题最常见。原因通常是规则阈值和相似度阈值设置不合理。建议的排查顺序检查输入数据是否清洗干净表情和 URL 是否残留。查看命中的红旗规则是否为误报。检查相似度阈值的分布收集 50 条正样本和负样本计算最佳阈值。不要只看总体分数要把每个子模块的结果单独输出便于定位问题。8.3 数据合规风险这可能是整个项目中最需要重视的问题。处理聊天记录、社交关系数据时必须注意获得用户明确授权说明数据用途和处理方式。建议对数据进行匿名化处理去除真实姓名、电话号码、住址等敏感信息。处理完成后及时清理中间数据不保留原始记录超过业务必要时间。不要使用爬虫大规模采集第三方平台数据这会涉及法律风险。评估结果只能作为“辅助参考”不能作为公开评价、人身攻击的依据。如果你的项目要上线请务必咨询法律专业人士确保符合当地数据保护法规。9. 最佳实践与工程建议9.1 规则引擎与模型结合我在实际项目中的经验是规则引擎负责“稳定召回”模型负责“复杂语义理解”。先把规则能覆盖的典型场景做扎实再逐步引入模型。这样做既降低成本也方便在出问题时快速回溯。在本文示例中红旗信号检测就是典型的规则引擎语义相似度属于模型或向量检索部分。两者形成互补。9.2 数据质量优先垃圾进垃圾出。这句话在社交分析项目中体现得特别明显。建议在数据接入层做三件事去重同一用户多次输入的聊天记录需要合并去重。时间标准化统一时间格式便于做频率分析。完整性评估计算数据覆盖率比如“该对象 90% 的消息都有记录”和“只有 10% 的记录”可信度完全不同。9.3 生产环境需要关注的边界并发控制FastAPI 默认同步执行如果模型推理很慢建议把推理任务放到队列中异步执行。缓存对相同的 embedding 结果做缓存减少模型调用。监控记录每次评估的分数分布当分数分布异常偏移时及时检查是数据问题还是规则被绕过。人工审核高风险判断建议加入人工复核流程避免算法误判造成恶劣影响。模型更新规则库和模型需要持续迭代。你可以把用户反馈接入系统用反馈样本重新训练或调阈值。9.4 关于“AI 相亲”类产品的定位最后想分享一点工程之外的体会。AI 在相亲场景中能做的事情是降低信息不对称帮助用户发现潜在风险信号。但它不能替代真实交往中的沟通和判断。作为开发者在设计这类系统时应该在界面上始终提醒用户“AI 评估结果仅供参考不构成对他人人格的定论。”这既是技术边界也是产品责任的底线。