AI相亲辅助系统：用NLP与关系网络识别海王

1. 为什么需要“AI 相亲，专业屏蔽海王”？

最近在和朋友聊到社交软件时，大家都有一个很直接的感受：相亲交友类 App 越来越多，但遇到“海王”的概率并没有下降。所谓“海王”，通常指那些同时维持多条暧昧关系、聊天话术高度标准化、关系推进节奏异常熟练的社交对象。对于认真想找长期关系的人来说，这类对象非常消耗精力和情绪。

我们能不能用 AI 技术来辅助判断？答案是肯定的。虽然 AI 无法百分之百判断一个人的真实人品，但可以从公开的社交行为数据、聊天文本、资料信息、互动频率等维度，构建一套“关系风险识别”机制。这种机制不是玄学，而是建立在自然语言处理、文本相似度、关系网络分析和规则引擎之上的工程系统。

本文将从工程实践角度，完整拆解一个“AI 相亲辅助筛选系统”的设计思路、模块拆分、代码实现和部署要点。无论你是想做社交产品，还是想自己写一套工具用于自我保护，都可以从这篇文章中找到可以直接使用的参考方案。

读完本文，你会掌握：

• 如何用文本分类和规则引擎检测聊天中的“海王话术”。
• 如何用向量检索判断一个人是否对多个对象使用高度相似的资料描述。
• 如何用关系网络分析识别“一对多”的社交连接模式。
• 如何把上述能力整合成一个可运行的 FastAPI 服务。
• 如何避免算法误判和数据合规风险。

2. 整体技术方案与模块拆分

2.1 系统架构总览

首先明确：这不是一个“一键识别渣男”的魔法系统，而是一个多模块协同的风险辅助判断系统。它更像一个“社交关系风控引擎”。

从功能角度看，系统可以拆成四层：

• 数据接入层：获取聊天记录、个人资料、社交关系列表等数据。
• 特征分析层：对文本内容、资料内容、关系结构进行特征提取。
• 风险决策层：结合规则模型、语义模型、网络模型输出风险评分。
• 展示输出层：以可视化报告的形式呈现风险信号和综合建议。

这个分层思路适合大多数社交风控项目。即使你现在只做一个小工具，也建议按这个思路组织代码，否则后续扩展会非常痛苦。

2.2 核心模块职责

模块	主要职责	技术方向
文本采集与预处理	清洗聊天记录、分词、去除表情和冗余信息	Python、正则表达式
红旗信号检测	识别海王高频话术、pua 式表达、快速推进关系等特征	规则引擎、文本分类
资料相似度分析	检测对多个对象使用相似自我介绍、相似问候语	向量检索、余弦相似度
关系网络分析	分析一对多连接、单向联系频率、时间分布	NetworkX、图数据库
风险评分引擎	融合多维特征输出综合风险指数	加权评分、逻辑回归
报告生成接口	输出可读的评估报告	FastAPI、前端模板

2.3 推荐技术选型

这里给出一个常见技术栈。你可以根据实际环境替换版本，但整体思路保持一致：

• 编程语言：Python 3.9+
• Web 框架：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(r"http\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 向量检索在社交筛选中的作用

红旗规则只能检测“话术是否可疑”，但有一个更隐蔽的场景：一个人同时对多个相亲对象发送几乎相同的自我介绍和问候。这种情况靠关键词规则很难发现，因为文字可能换了一种说法，但语义保持一致。

向量相似度分析可以解决这个问题。我们可以把一段文本编码成一个高维向量，然后计算两段文本之间的余弦相似度。相似度越高，说明语义越接近。

在实际系统中，常见的做法是：

1. 收集目标对象的自我介绍、开场白、日常聊天常用句。
2. 将所有文本编码为向量。
3. 对新的聊天对象，计算新文本与历史文本的相似度。
4. 如果相似度超过阈值，则提示“存在复用话术的可能性”。

5.2 基于 sentence-transformers 的实现

这里给出一个通用实现。使用sentence-transformers库可以加载开源 embedding 模型。

# 文件路径：modules/embedding.py import numpy as np class TextEmbedding: def __init__(self, model_name="paraphrase-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_embeddings=True) # 降级方案：使用简单的 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_pattern=r"\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, axis=1).reshape(-1, 1) norm2 = np.linalg.norm(vec2, axis=1).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_count=item.get("msg_count", 0), last_time=item.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(title="AI 相亲风险辅助评估系统") 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_model=AssessResponse) 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, host="0.0.0.0", port=8000)

这里的关键是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-v2
FAISS 安装失败	编译环境问题	使用faiss-cpu版本；或者改用 numpy 手写余弦相似度

8.2 识别不准确

这个问题最常见。原因通常是规则阈值和相似度阈值设置不合理。

建议的排查顺序：

1. 检查输入数据是否清洗干净，表情和 URL 是否残留。
2. 查看命中的红旗规则是否为误报。
3. 检查相似度阈值的分布，收集 50 条正样本和负样本计算最佳阈值。
4. 不要只看总体分数，要把每个子模块的结果单独输出，便于定位问题。

8.3 数据合规风险

这可能是整个项目中最需要重视的问题。

处理聊天记录、社交关系数据时，必须注意：

• 获得用户明确授权，说明数据用途和处理方式。
• 建议对数据进行匿名化处理，去除真实姓名、电话号码、住址等敏感信息。
• 处理完成后及时清理中间数据，不保留原始记录超过业务必要时间。
• 不要使用爬虫大规模采集第三方平台数据，这会涉及法律风险。
• 评估结果只能作为“辅助参考”，不能作为公开评价、人身攻击的依据。

如果你的项目要上线，请务必咨询法律专业人士，确保符合当地数据保护法规。

9. 最佳实践与工程建议

9.1 规则引擎与模型结合

我在实际项目中的经验是：规则引擎负责“稳定召回”，模型负责“复杂语义理解”。先把规则能覆盖的典型场景做扎实，再逐步引入模型。这样做既降低成本，也方便在出问题时快速回溯。

在本文示例中，红旗信号检测就是典型的规则引擎；语义相似度属于模型或向量检索部分。两者形成互补。

9.2 数据质量优先

垃圾进，垃圾出。这句话在社交分析项目中体现得特别明显。

建议在数据接入层做三件事：

• 去重：同一用户多次输入的聊天记录需要合并去重。
• 时间标准化：统一时间格式，便于做频率分析。
• 完整性评估：计算数据覆盖率，比如“该对象 90% 的消息都有记录”和“只有 10% 的记录”，可信度完全不同。

9.3 生产环境需要关注的边界

• 并发控制：FastAPI 默认同步执行，如果模型推理很慢，建议把推理任务放到队列中异步执行。
• 缓存：对相同的 embedding 结果做缓存，减少模型调用。
• 监控：记录每次评估的分数分布，当分数分布异常偏移时，及时检查是数据问题还是规则被绕过。
• 人工审核：高风险判断建议加入人工复核流程，避免算法误判造成恶劣影响。
• 模型更新：规则库和模型需要持续迭代。你可以把用户反馈接入系统，用反馈样本重新训练或调阈值。

9.4 关于“AI 相亲”类产品的定位

最后想分享一点工程之外的体会。AI 在相亲场景中能做的事情，是降低信息不对称，帮助用户发现潜在风险信号。但它不能替代真实交往中的沟通和判断。

作为开发者，在设计这类系统时，应该在界面上始终提醒用户：“AI 评估结果仅供参考，不构成对他人人格的定论。”这既是技术边界，也是产品责任的底线。