尧图网站建设 
线上相亲交友场景里“海王”这个词指的是那些同时与多人保持暧昧、使用套路化话术、回避承诺的聊天对象。真正想做一个“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依赖用途说明如下依赖用途fastapi提供 HTTP 接口uvicorn启动异步服务pydantic做请求参数校验和响应模型pandas 和 scikit-learn训练和加载轻量分类模型joblib保存和加载模型文件openai调用 OpenAI 兼容的大模型接口如果原始环境没有 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, threshold0.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风险等级映射表如下风险分风险等级前端提示示例0 - 39low目前未发现明显海王特征可以继续了解40 - 69medium部分行为值得留意建议多确认对方真实信息70 - 100high存在较多套路化聊天特征建议提高警惕注意规则阈值需要根据真实聊天语料调参。上面代码里的阈值只适用于原型不能直接作为生产标准。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_features5000), LogisticRegression(max_iter1000), ) 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_keyos.getenv(LLM_API_KEY, local), base_urlos.getenv(LLM_BASE_URL, http://localhost:11434/v1), timeout15, ) 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( modelself.model, messages[{role: user, content: prompt}], max_tokens500, ) 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_pathNone, llm_analyzerNone): 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_llmTrue): 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_idconversation.conversation_id, risk_scorefinal_score, risk_levellevel, risk_tagstags, hit_ruleshit_rules, suggestionssuggestions, raw_featuresfeatures, )整个融合过程遵循两个原则大模型只负责补充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(titleAI 相亲风险提示服务, version0.1.0) # 初始化时可以先不传 model_path便于本地快速跑通 assessor RiskAssessor(model_pathNone, llm_analyzerLLMAnalyzer()) app.get(/health) def health(): return {status: ok} app.post(/api/v1/assess, response_modelRiskReport) def assess(conv: Conversation): if not conv.messages: raise HTTPException(status_code400, detailmessages 不能为空) if len(conv.messages) 500: raise HTTPException(status_code400, detailmessages 过长最多 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_idtest, messages[Message(senders, contentc) for s, c in messages], ) def test_high_risk_sample(): assessor RiskAssessor(model_pathNone) conv build_conv([ (me, 你周末一般都做什么), (other, 宝贝想你了么么哒), (me, 你是做什么工作的呀), (other, 这个不重要啦以后再说嘛), (me, 我们聊了挺久你觉得我怎么样), (other, 你很好呀看缘分嘛顺其自然), ]) report assessor.analyze(conv, use_llmFalse) assert report.risk_score 40 assert report.risk_level in (medium, high) def test_empty_conversation(): assessor RiskAssessor(model_pathNone) conv build_conv([]) try: assessor.analyze(conv, use_llmFalse) 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 适合学习和功能验证距离生产还需要补齐以下能力模块当前状态生产要求配置环境变量零散统一配置中心或环境变量文件日志默认输出输出结构化 JSON 日志包含请求 ID监控无增加接口延迟、成功率、风险等级分布指标权限无增加 API Key 鉴权限制调用频率数据存储不落库评估是否存储脱敏后的聊天记录回滚无模型文件版本化管理支持按版本回滚生产环境不要直接在公网部署无鉴权接口。相亲聊天记录涉及个人隐私接口必须放在登录之后并记录审计日志。7.2 隐私、权限与合规设计聊天记录属于高度敏感数据设计时要遵循最小化原则只上传必要的聊天片段而不是全部聊天记录。本地先做脱敏把手机号、微信号、具体地址替换成占位符。大模型调用前发送脱敏文本避免隐私泄露。用户应明确知晓“AI 风险提示”功能并具有关闭权限。不建议在数据库中明文保存原始聊天记录。如果业务面向真实用户需要咨询法务和合规人员确认数据处理范围、留存期限和用户授权方式。7.3 从单个接口扩展到 AI Agent 和多模态识别当前实现是一个“分析一次聊天记录”的接口。进一步可以扩展成 AI Agent 工作流定时分析新消息当风险分升高时自动通知用户。把语音转文本、图片转文字结果统一并入分析。结合用户反馈把“这个人是海王”的举报结果变成新的训练样本。将规则引擎和模型评分结果写入特征日志用于持续调优。从开发节奏来看不建议一开始就上 Agent。先把规则版跑通再接入大模型最后再做成自动触发的 Agent 服务。每一步都要有日志和回归测试。7.4 新手学习路径如果正在学习 AI 应用开发或 AI 工程实践这个项目是很好的练手素材。建议按以下顺序推进先跑通规则引擎理解“把行为转为特征”。再增加 TF-IDF 和逻辑回归理解机器学习建模流程。接入大模型练习 prompt 设计和结构化输出解析。用 FastAPI 封装接口熟悉前后端联调。最后补充测试、日志、监控和部署完成工程闭环。每一步都可以独立验证不必一次性做完。项目代码始终以“能解释、能定位、可回滚”为目标这比堆叠模型更重要。 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依赖用途说明如下依赖用途fastapi提供 HTTP 接口uvicorn启动异步服务pydantic做请求参数校验和响应模型pandas 和 scikit-learn训练和加载轻量分类模型joblib保存和加载模型文件openai调用 OpenAI 兼容的大模型接口如果原始环境没有 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, threshold0.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风险等级映射表如下风险分风险等级前端提示示例0 - 39low目前未发现明显海王特征可以继续了解40 - 69medium部分行为值得留意建议多确认对方真实信息70 - 100high存在较多套路化聊天特征建议提高警惕注意规则阈值需要根据真实聊天语料调参。上面代码里的阈值只适用于原型不能直接作为生产标准。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_features5000), LogisticRegression(max_iter1000), ) 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_keyos.getenv(LLM_API_KEY, local), base_urlos.getenv(LLM_BASE_URL, http://localhost:11434/v1), timeout15, ) 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( modelself.model, messages[{role: user, content: prompt}], max_tokens500, ) 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_pathNone, llm_analyzerNone): 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_llmTrue): 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_idconversation.conversation_id, risk_scorefinal_score, risk_levellevel, risk_tagstags, hit_ruleshit_rules, suggestionssuggestions, raw_featuresfeatures, )整个融合过程遵循两个原则大模型只负责补充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(titleAI 相亲风险提示服务, version0.1.0) # 初始化时可以先不传 model_path便于本地快速跑通 assessor RiskAssessor(model_pathNone, llm_analyzerLLMAnalyzer()) app.get(/health) def health(): return {status: ok} app.post(/api/v1/assess, response_modelRiskReport) def assess(conv: Conversation): if not conv.messages: raise HTTPException(status_code400, detailmessages 不能为空) if len(conv.messages) 500: raise HTTPException(status_code400, detailmessages 过长最多 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_idtest, messages[Message(senders, contentc) for s, c in messages], ) def test_high_risk_sample(): assessor RiskAssessor(model_pathNone) conv build_conv([ (me, 你周末一般都做什么), (other, 宝贝想你了么么哒), (me, 你是做什么工作的呀), (other, 这个不重要啦以后再说嘛), (me, 我们聊了挺久你觉得我怎么样), (other, 你很好呀看缘分嘛顺其自然), ]) report assessor.analyze(conv, use_llmFalse) assert report.risk_score 40 assert report.risk_level in (medium, high) def test_empty_conversation(): assessor RiskAssessor(model_pathNone) conv build_conv([]) try: assessor.analyze(conv, use_llmFalse) 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 适合学习和功能验证距离生产还需要补齐以下能力模块当前状态生产要求配置环境变量零散统一配置中心或环境变量文件日志默认输出输出结构化 JSON 日志包含请求 ID监控无增加接口延迟、成功率、风险等级分布指标权限无增加 API Key 鉴权限制调用频率数据存储不落库评估是否存储脱敏后的聊天记录回滚无模型文件版本化管理支持按版本回滚生产环境不要直接在公网部署无鉴权接口。相亲聊天记录涉及个人隐私接口必须放在登录之后并记录审计日志。7.2 隐私、权限与合规设计聊天记录属于高度敏感数据设计时要遵循最小化原则只上传必要的聊天片段而不是全部聊天记录。本地先做脱敏把手机号、微信号、具体地址替换成占位符。大模型调用前发送脱敏文本避免隐私泄露。用户应明确知晓“AI 风险提示”功能并具有关闭权限。不建议在数据库中明文保存原始聊天记录。如果业务面向真实用户需要咨询法务和合规人员确认数据处理范围、留存期限和用户授权方式。7.3 从单个接口扩展到 AI Agent 和多模态识别当前实现是一个“分析一次聊天记录”的接口。进一步可以扩展成 AI Agent 工作流定时分析新消息当风险分升高时自动通知用户。把语音转文本、图片转文字结果统一并入分析。结合用户反馈把“这个人是海王”的举报结果变成新的训练样本。将规则引擎和模型评分结果写入特征日志用于持续调优。从开发节奏来看不建议一开始就上 Agent。先把规则版跑通再接入大模型最后再做成自动触发的 Agent 服务。每一步都要有日志和回归测试。7.4 新手学习路径如果正在学习 AI 应用开发或 AI 工程实践这个项目是很好的练手素材。建议按以下顺序推进先跑通规则引擎理解“把行为转为特征”。再增加 TF-IDF 和逻辑回归理解机器学习建模流程。接入大模型练习 prompt 设计和结构化输出解析。用 FastAPI 封装接口熟悉前后端联调。最后补充测试、日志、监控和部署完成工程闭环。每一步都可以独立验证不必一次性做完。项目代码始终以“能解释、能定位、可回滚”为目标这比堆叠模型更重要。