AI相亲聊天风险识别：融合规则、机器学习与大模型的实战方案

线上相亲交友场景里，“海王”这个词指的是那些同时与多人保持暧昧、使用套路化话术、回避承诺的聊天对象。真正想做一个“AI相亲，专业屏蔽海王”的小工具，不只是训练一个能输出“是/否”的分类模型，而是要把“海王”行为拆成可解释的文本特征，再通过规则、统计模型和大模型分析给用户一个风险提示。本文围绕这个目标，从零实现一个相亲聊天风险提示服务。读者会看到完整的环境准备、项目结构、规则引擎、机器学习模型、大模型点评、FastAPI 接口、测试与排查方式，最终能复现一个可以在本机跑起来的 Demo。

这个项目在工程上有一个很现实的问题：完全靠人工读聊天记录判断效率太低，完全靠大模型判断又会出现“答非所问”“格式不稳定”“成本不可控”。因此本文采用规则引擎兜底、统计特征辅助、大模型生成解释的混合方案。这样既能在离线场景快速给出可解释的结果，又能在接入大模型后提升提示语的自然度。整个项目用 Python 编写，前后端通过 JSON 交互，便于迁移到小程序、Web 或 App 后端。

1. 先理解“海王识别”的本质是文本风险识别

1.1 把“海王”行为变成可计算特征

“海王”不是一个严谨的技术概念，而是一类社交行为模式。要做成程序，就必须先把这类行为翻译成可以计算信号。结合相亲聊天场景，常见的风险信号包括以下几类：

• 群发和复制粘贴：同一句问候或自我介绍发给不同对象，聊天记录中出现大量重复模板。
• 信息收敛：很少介绍自己的真实情况，不回答工作、住址、家庭等具体问题。
• 回避承诺：提到关系、见面、未来规划时经常用“看缘分”“顺其自然”“以后再说”等回应。
• 高频夸张赞美：认识时间很短，就频繁使用“宝贝”“亲爱的小可爱”等称呼，或不断夸对方外貌。
• 时间线矛盾：活跃时间段跨度极大，凌晨也在频繁回复，疑似同时维护多个对话。
• 投入度不平衡：大量短消息、表情包、问候语，但很少提出深入问题，也不认真回答对方问题。

这些信号对应到 NLP 任务上，分别是重复文本检测、关键词规则、提问比例统计、称呼褒贬统计、时间戳分布计算。它们并不需要特别复杂的模型，先用规则处理，能覆盖大部分明显案例。

1.2 技术方案：规则、统计模型和大模型互补

在技术选型上，不建议只押注某一种方案。规则引擎可解释性强，适合快速拦截“撒网式开场白”；统计模型能够学习更复杂的表达方式，比如“抄作业式回复”的不同变体；大模型则擅长总结对话脉络，生成给人看的风险建议。

三种方案的对比关系如下：

实际工程中，规则引擎负责“为什么扣分”，统计模型负责“整体概率”，大模型负责“说什么”。三者融合后，风险报告既包含分数，也包含理由，用户更容易理解。

1.3 风险提示与应用边界

“AI相亲，专业屏蔽海王”里的“屏蔽”不建议做成系统自动拉黑，而是做成风险提示与确认动作。因为聊天记录只能反映文本内容，无法判断完整事实。一个用户可能因为表达能力差，被规则误判为“回避问题”；也可能确实在同时接触多个对象，但并没有欺骗意图。

因此工程上要做两件事：

• 输出风险等级和具体命中规则，让用户看到判断依据。
• 只提供“建议保持距离”“建议进一步确认”等提示，不输出“这个人一定是海王”的结论。

系统边界设置清楚后，才能减少误伤，也才符合产品合规要求。

2. 搭建“AI相亲风险提示”最小工程

2.1 环境准备与依赖

本项目的运行环境比较简单，Python 3.10 以上即可。先创建独立虚拟环境，避免依赖冲突。

mkdir ai-match-risk cd ai-match-risk python -m venv venv source venv/bin/activate pip install --upgrade pip

项目依赖按模块拆分：

pip install fastapi uvicorn pydantic pandas scikit-learn joblib python-dotenv

下面这些依赖用于可选能力：

pip install openai

如果要用本地分词增强中文特征，可以额外安装：

pip install jieba

依赖用途说明如下：

如果原始环境没有 GPU，模型部分不要使用 BERT 等重模型，先用 TF-IDF 加逻辑回归跑通流程。

2.2 项目目录结构与数据约定

推荐目录结构如下：

ai-match-risk/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── models.py │ ├── rules.py │ ├── features.py │ ├── assessor.py │ └── llm_analyzer.py ├── scripts/ │ └── train_model.py ├── tests/ │ └── test_assessor.py ├── requirements.txt └── chat_samples.json

其中app/models.py定义请求和响应结构，rules.py实现规则引擎，features.py负责从聊天记录里抽取特征，assessor.py汇总规则、模型和大模型结果，main.py是 FastAPI 入口。

聊天记录统一使用conversation_id和messages数组。每个消息包含sender、content、timestamp三个字段。在原型阶段，约定发送方只有两种取值：me代表当前用户，other代表对方。

2.3 定义接口数据结构

在app/models.py中定义协议如下：

from typing import List, Optional from datetime import datetime from pydantic import BaseModel, Field class Message(BaseModel): sender: str = Field(..., description="消息发送方，me 或 other") content: str = Field(..., description="消息文本内容") timestamp: Optional[datetime] = Field(None, description="消息时间") class Conversation(BaseModel): conversation_id: str messages: List[Message] class RiskReport(BaseModel): conversation_id: str risk_score: float risk_level: str risk_tags: List[str] hit_rules: List[str] suggestions: List[str] raw_features: dict

这里的关键是让接口返回结构稳定。即使后续算法变了，前端也可以依赖risk_score和risk_level渲染结果。

3. 实现规则引擎：先让程序能抓住明显套路

3.1 重复内容检测：识别复制粘贴与群发

群发是“海王”最常见的行为特征。同一个用户在不同时间点发完全相同或高度相似的句子，大概率是复制粘贴开场白。

在app/features.py中实现基于difflib的重复检测：

from difflib import SequenceMatcher from collections import defaultdict def detect_repeated_content(messages, threshold=0.85): sender_messages = defaultdict(list) for msg in messages: if msg.sender == "other": sender_messages[msg.sender].append(msg.content.strip()) repeat_count = 0 texts = sender_messages["other"] for i in range(len(texts)): for j in range(i + 1, len(texts)): if len(texts[i]) < 6 or len(texts[j]) < 6: continue ratio = SequenceMatcher(None, texts[i], texts[j]).ratio() if ratio >= threshold: repeat_count += 1 return repeat_count

这段代码对“对方”发送的内容两两比较，相似度超过 0.85 就认为存在重复。重复次数保留为特征值，不直接判定，避免因为偶尔回一句相同的话而误伤。

3.2 统计特征：衡量投入程度和聊天健康度

在app/features.py中增加统计特征提取函数。这里需要处理中英文混合文本，并计算以下几类特征：

import re def compute_conversation_features(messages): if not messages: return {} other_msgs = [m for m in messages if m.sender == "other"] if not other_msgs: return {} other_count = len(other_msgs) total_len = sum(len(m.content) for m in other_msgs) avg_len = total_len / other_count question_count = sum( 1 for m in other_msgs if re.search(r"[?？]$", m.content.strip()) ) question_ratio = question_count / other_count short_count = sum(1 for m in other_msgs if len(m.content.strip()) < 8) short_msg_ratio = short_count / other_count praise_words = ["宝贝", "亲爱", "小可爱", "你真好看", "么么哒", "宝宝"] praise_count = sum( m.content.count(word) for m in other_msgs for word in praise_words ) avoid_words = ["以后再说", "看缘分", "顺其自然", "不想谈这个", "到时候看"] avoid_count = sum( m.content.count(word) for m in other_msgs for word in avoid_words ) timestamps = [m.timestamp for m in other_msgs if m.timestamp] active_span = 0 if len(timestamps) >= 2: hours = [(t.hour + t.minute / 60) for t in timestamps] active_span = max(hours) - min(hours) return { "other_message_count": other_count, "other_avg_len": round(avg_len, 2), "other_question_ratio": round(question_ratio, 2), "other_short_msg_ratio": round(short_msg_ratio, 2), "other_praise_count": praise_count, "avoid_commitment_count": avoid_count, "active_time_span_hours": round(active_span, 2), "repeated_content_times": detect_repeated_content(messages), }

这里的统计口径会影响评分结果。例如question_ratio使用以问号结尾的句子数量占比，避免把“你是不是叫 XX”这类反问句排除在外。short_msg_ratio衡量的是碎片化表达程度。

3.3 规则评分与风险等级映射

规则引擎不追求每一项都准确，而是把命中项累加成风险分。在app/rules.py中实现如下：

class RuleEngine: def evaluate(self, f: dict): score = 0 hit_rules = [] if f.get("repeated_content_times", 0) >= 2: score += 20 hit_rules.append("同一句话在不同时间重复出现，疑似模板群发") if ( f.get("other_message_count", 0) >= 10 and f.get("other_question_ratio", 0) < 0.12 ): score += 15 hit_rules.append("很少主动提问，互动停留在客套层") if f.get("other_short_msg_ratio", 0) >= 0.6: score += 10 hit_rules.append("大量短消息，深度交流不足") if f.get("other_praise_count", 0) >= 3: score += 15 hit_rules.append("高频夸张赞美，存在套路化讨好嫌疑") if f.get("avoid_commitment_count", 0) >= 1: score += 15 hit_rules.append("多次回避关系相关话题") if ( f.get("other_message_count", 0) >= 30 and f.get("active_time_span_hours", 0) >= 10 ): score += 10 hit_rules.append("活跃时间段跨度大，可能同时维护多个对话") score = min(100, score) if score >= 70: level = "high" elif score >= 40: level = "medium" else: level = "low" return round(score, 2), hit_rules, level

风险等级映射表如下：

注意：规则阈值需要根据真实聊天语料调参。上面代码里的阈值只适用于原型，不能直接作为生产标准。

4. 加入机器学习与大模型风险点评

4.1 用 TF-IDF 加逻辑回归训练轻量分类器

规则引擎能解释“为什么”，但无法覆盖新表达。可以收集一批标注数据，训练一个轻量分类器。在scripts/train_model.py中准备最小训练脚本：

import pandas as pd import joblib from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.linear_model import LogisticRegression from sklearn.pipeline import make_pipeline # 示例数据，正式使用需要扩充并人工标注 data = [ ("今天忙什么呀？我刚下班，有点累。", 0), ("宝贝想你了，么么哒，你在干嘛呢？", 1), ("我是做设计的，平时喜欢看展，你呢？", 0), ("我很少聊这些，以后再说吧。", 1), ("我们好像聊了挺久，你觉得我怎么样？", 0), ("你很好呀，看缘分嘛，顺其自然。", 1), ("周末去爬山吗？我认识一个不错的地方。", 0), ("宝宝你真好看，亲一个。", 1), ] df = pd.DataFrame(data, columns=["text", "label"]) model = make_pipeline( TfidfVectorizer(ngram_range=(1, 2), max_features=5000), LogisticRegression(max_iter=1000), ) model.fit(df["text"], df["label"]) joblib.dump(model, "risk_model.joblib") print("model saved")

训练脚本的关键点：

• 数据量小只能演示流程，不能代表真实效果。
• ngram_range=(1, 2)同时考虑单词和相邻词组。
• 逻辑回归训练快，适合快速验证特征是否有区分度。
• 样本需要均衡，否则模型会把绝大多数对话判成低风险。

4.2 调用大模型生成可解释建议

大模型在这里不是必需的，但可以显著提升提示语的自然度。为了兼容不同环境，在app/llm_analyzer.py中调用 OpenAI 兼容接口。本地也可以部署支持 OpenAI 协议的开源模型，用base_url指向本地服务。

import json import os from openai import OpenAI class LLMAnalyzer: def __init__(self): self.client = OpenAI( api_key=os.getenv("LLM_API_KEY", "local"), base_url=os.getenv("LLM_BASE_URL", "http://localhost:11434/v1"), timeout=15, ) self.model = os.getenv("LLM_MODEL", "qwen2.5:7b") def analyze(self, conversation, risk_score): sample_messages = "\n".join( f"{m.sender}: {m.content[:100]}" for m in conversation.messages[-20:] ) prompt = f""" 你是婚恋交友安全提示助手。下面是一段聊天记录。 请你判断其中是否存在“海王”式聊天特征，例如群发模板、回避承诺、夸大赞美、投入度不平衡等。 风险分已经是 {risk_score} 分，满分 100。 只输出 JSON，格式如下： {{"tags": [], "reasons": [], "suggestions": []}} 不要直接下结论说某人一定是海王，只做客观风险提示。 聊天记录： {sample_messages} """ response = self.client.chat.completions.create( model=self.model, messages=[{"role": "user", "content": prompt}], max_tokens=500, ) content = response.choices[0].message.content return json.loads(content)

使用大模型时需要注意：

• 提示词里明确要求“只输出 JSON”，方便解析。
• 截断聊天记录，避免超出模型上下文窗口。
• 设置timeout，避免接口长时间挂起。
• 不允许模型输出“一定海王”这类绝对结论。

4.3 规则、模型和大模型的融合方式

在app/assessor.py中实现融合逻辑：

import joblib from .rules import RuleEngine from .features import compute_conversation_features from .models import RiskReport class RiskAssessor: def __init__(self, model_path=None, llm_analyzer=None): self.rule_engine = RuleEngine() self.model_path = model_path self.model = None if model_path: self.model = joblib.load(model_path) self.llm_analyzer = llm_analyzer def _model_score(self, messages): if not self.model: return None text = " ".join(m.content for m in messages) prob = self.model.predict_proba([text])[0][1] return round(prob * 100, 2) def analyze(self, conversation, use_llm=True): features = compute_conversation_features(conversation.messages) rule_score, hit_rules, _ = self.rule_engine.evaluate(features) ml_score = self._model_score(conversation.messages) if ml_score is not None: final_score = round(0.7 * rule_score + 0.3 * ml_score, 2) else: final_score = rule_score if final_score >= 70: level = "high" elif final_score >= 40: level = "medium" else: level = "low" suggestions = [] if level == "high": suggestions.append("建议暂缓推进关系，优先确认对方真实身份信息") suggestions.append("避免向对方透露住址、财务状况等敏感信息") elif level == "medium": suggestions.append("建议多通过语音或视频确认对方身份") suggestions.append("观察对方是否愿意详细回答你的问题") else: suggestions.append("目前风险较低，但仍建议保护个人隐私") llm_result = None if use_llm and self.llm_analyzer: try: llm_result = self.llm_analyzer.analyze(conversation, final_score) except Exception as exc: llm_result = {"error": str(exc)} tags = [] if llm_result and "tags" in llm_result: tags = llm_result["tags"] elif level == "high": tags = ["模板化表达", "回避承诺"] return RiskReport( conversation_id=conversation.conversation_id, risk_score=final_score, risk_level=level, risk_tags=tags, hit_rules=hit_rules, suggestions=suggestions, raw_features=features, )

整个融合过程遵循两个原则：

• 大模型只负责补充tags和解释，不参与核心评分。
• 大模型失败时，规则引擎仍能返回结果，接口不降级失败。

5. 封装 FastAPI 接口并验证

5.1 服务类与接口定义

在app/main.py中创建 FastAPI 应用：

from fastapi import FastAPI, HTTPException from .models import Conversation, RiskReport from .assessor import RiskAssessor from .llm_analyzer import LLMAnalyzer app = FastAPI(title="AI 相亲风险提示服务", version="0.1.0") # 初始化时可以先不传 model_path，便于本地快速跑通 assessor = RiskAssessor(model_path=None, llm_analyzer=LLMAnalyzer()) @app.get("/health") def health(): return {"status": "ok"} @app.post("/api/v1/assess", response_model=RiskReport) def assess(conv: Conversation): if not conv.messages: raise HTTPException(status_code=400, detail="messages 不能为空") if len(conv.messages) > 500: raise HTTPException(status_code=400, detail="messages 过长，最多 500 条") return assessor.analyze(conv)

接口设计上做了两个保护：

• 空消息直接返回 400，避免规则引擎收到空数组。
• 超过 500 条消息直接拒绝，防止大模型 prompt 过长和接口耗时失控。

5.2 启动服务和用 curl 调用

启动服务：

uvicorn app.main:app --reload --port 8000

用一个高风险样例测试：

curl -X POST http://127.0.0.1:8000/api/v1/assess \ -H "Content-Type: application/json" \ -d '{ "conversation_id": "demo_high", "messages": [ {"sender": "me", "content": "你周末一般都做什么？", "timestamp": "2025-01-01T10:00:00"}, {"sender": "other", "content": "宝贝想你了，么么哒", "timestamp": "2025-01-01T10:00:03"}, {"sender": "me", "content": "你是做什么工作的呀？", "timestamp": "2025-01-01T10:00:30"}, {"sender": "other", "content": "这个不重要啦，以后再说嘛", "timestamp": "2025-01-01T10:00:35"}, {"sender": "me", "content": "我们聊了挺久，你觉得我怎么样？", "timestamp": "2025-01-01T10:01:00"}, {"sender": "other", "content": "你很好呀，看缘分嘛，顺其自然", "timestamp": "2025-01-01T10:01:05"}, {"sender": "me", "content": "你之前也是这么回别人的吗？", "timestamp": "2025-01-01T10:02:00"}, {"sender": "other", "content": "宝贝你又乱想了，么么哒", "timestamp": "2025-01-01T10:02:03"} ] }'

如果大模型不可用，接口会在llm_result中记录错误，但核心响应仍能返回。

5.3 自动化测试与回归保障

在tests/test_assessor.py中写两个最小测试：

from app.assessor import RiskAssessor from app.models import Conversation, Message def build_conv(messages): return Conversation( conversation_id="test", messages=[Message(sender=s, content=c) for s, c in messages], ) def test_high_risk_sample(): assessor = RiskAssessor(model_path=None) conv = build_conv([ ("me", "你周末一般都做什么？"), ("other", "宝贝想你了，么么哒"), ("me", "你是做什么工作的呀？"), ("other", "这个不重要啦，以后再说嘛"), ("me", "我们聊了挺久，你觉得我怎么样？"), ("other", "你很好呀，看缘分嘛，顺其自然"), ]) report = assessor.analyze(conv, use_llm=False) assert report.risk_score >= 40 assert report.risk_level in ("medium", "high") def test_empty_conversation(): assessor = RiskAssessor(model_path=None) conv = build_conv([]) try: assessor.analyze(conv, use_llm=False) except ValueError: pass

需要根据实际实现调整空消息的异常处理。如果compute_conversation_features对空数组返回空字典，应在analyze开头加空消息校验。

6. 常见问题与排查路径

6.1 中文文本特征效果差

现象：规则引擎能跑，但统计模型给出的风险分不稳定，甚至把明显正常对话判成高风险。

可能原因：

• 没有做中文分词，直接按空格切分导致 TF-IDF 特征稀疏。
• 关键词规则覆盖不足，命中太少。
• 训练样本太少，且类别不均衡。

排查方式：

python -c "from app.features import compute_conversation_features; print(compute_conversation_features([]))"

解决方案：

• 引入jieba做分词，并把分词结果用空格连接后再做 TF-IDF。
• 增加训练样本，让正负样本比例接近 1:1。
• 先看规则引擎命中了哪些规则，再决定是否需要调整阈值。

6.2 大模型调用超时或返回格式错误

现象：接口响应很慢，或解析json.loads时抛出JSONDecodeError。

可能原因：

• 大模型服务未启动，或base_url配错。
• 模型上下文太长，响应时间超过timeout。
• 模型没有严格按提示词输出 JSON，夹带了额外文字。

排查方式：

• 单独调用curl测试大模型接口是否能通。
• 打印 prompt 和 response 原文，确认格式。
• 降低max_tokens，只要求模型输出短结果。

解决方案：

• 在analyze方法里增加try except，解析失败时返回空标签。
• 把timeout从 15 秒改为 30 秒，但不要设置太长。
• 对输出做正则清洗，只保留 JSON 片段。

6.3 规则误报过多

现象：普通用户只是话少，结果被判定为高风险。

可能原因：

• question_ratio阈值过低，导致“话少但正常”被误判。
• 短消息比例过高，但对方可能只是不擅长打字。
• 赞美词表包含日常用词，比如“亲爱”出现在正常称呼中。

排查方式：

• 打印raw_features，查看具体哪个特征触发阈值。
• 单测里加入“低风险”样例，回归时验证不误报。

解决方案：

• 提高阈值，比如把question_ratio阈值从 0.12 降到 0.08。
• 增加最少消息条数条件，聊天不足 20 条不启用部分规则。
• 引入人工复核，在风险等级为 high 时标记为“待人工确认”。

6.4 生产环境模型加载与内存问题

现象：本地跑通，但部署到服务器后报FileNotFoundError或内存不足。

可能原因：

• 模型文件路径写死，生产目录结构和本地不一致。
• 使用过大模型文件，超出容器内存限制。

解决方案：

• 将模型路径放到环境变量，例如RISK_MODEL_PATH=/data/models/risk_model.joblib。
• 使用os.getenv读取路径，而不是硬编码相对路径。
• 如果内存有限，优先使用 TF-IDF + 逻辑回归，不要加载 BERT 类模型。

7. 工程化实践与扩展方向

7.1 从 Demo 到生产的改造清单

当前 Demo 适合学习和功能验证，距离生产还需要补齐以下能力：

生产环境不要直接在公网部署无鉴权接口。相亲聊天记录涉及个人隐私，接口必须放在登录之后，并记录审计日志。

7.2 隐私、权限与合规设计

聊天记录属于高度敏感数据，设计时要遵循最小化原则：

• 只上传必要的聊天片段，而不是全部聊天记录。
• 本地先做脱敏，把手机号、微信号、具体地址替换成占位符。
• 大模型调用前发送脱敏文本，避免隐私泄露。
• 用户应明确知晓“AI 风险提示”功能，并具有关闭权限。
• 不建议在数据库中明文保存原始聊天记录。

如果业务面向真实用户，需要咨询法务和合规人员，确认数据处理范围、留存期限和用户授权方式。

7.3 从单个接口扩展到 AI Agent 和多模态识别

当前实现是一个“分析一次聊天记录”的接口。进一步可以扩展成 AI Agent 工作流：

• 定时分析新消息，当风险分升高时自动通知用户。
• 把语音转文本、图片转文字结果统一并入分析。
• 结合用户反馈，把“这个人是海王”的举报结果变成新的训练样本。
• 将规则引擎和模型评分结果写入特征日志，用于持续调优。

从开发节奏来看，不建议一开始就上 Agent。先把规则版跑通，再接入大模型，最后再做成自动触发的 Agent 服务。每一步都要有日志和回归测试。

7.4 新手学习路径

如果正在学习 AI 应用开发或 AI 工程实践，这个项目是很好的练手素材。建议按以下顺序推进：

1. 先跑通规则引擎，理解“把行为转为特征”。
2. 再增加 TF-IDF 和逻辑回归，理解机器学习建模流程。
3. 接入大模型，练习 prompt 设计和结构化输出解析。
4. 用 FastAPI 封装接口，熟悉前后端联调。
5. 最后补充测试、日志、监控和部署，完成工程闭环。

每一步都可以独立验证，不必一次性做完。项目代码始终以“能解释、能定位、可回滚”为目标，这比堆叠模型更重要。