基于大模型与FastAPI的PUA操控话术识别系统实现

有些事放到技术语境里看会变得特别有意思。一个教别人用话术操控关系的“大师”他总结的那套“打压-拉扯-制造焦虑-推拉配合”的操作流程本质上是一套可以被穷举的文本模式。以前识别这种模式要靠人肉判断一条一条看聊天记录、看社群发言、看直播话术效率极低。但现在不一样了这套模式喂给大模型再用 Prompt 约束输出就能做到批量识别、打标签、生成反制建议甚至直接封装成 API 接进审核系统。这篇文章就把这套“反向工具”完整拆一遍怎么用自己的聊天记录或公开合规语料构造样本库怎么用本地部署的开源大模型做语义分类怎么用 FastAPI 把识别能力封装成接口以及怎么跑批量任务、观察显存占用、排查常见问题。这里说的“PUA 识别”指的是识别操控式沟通话术用于情感安全、内容审核、合规质检等方向不是教人怎么实施话术。整套方案的核心是把模糊的“感觉被拿捏”变成可量化的标签让套路本身变成可被程序识别的数据。适合看这篇文章的读者主要有三类第一类是 NLP 或 AI 应用工程师想找一个真实场景练手把大模型接进业务系统第二类是社群运营、内容安全、客服质检相关岗位需要自动筛选高风险对话第三类是产品经理或独立开发者想快速搭一个 AI 工具 Demo验证“AI 内容安全”方向的产品价值。下面直接进正文先看整体能力设计再一步步从零搭建。1. 核心能力速览能力项说明项目定位基于大模型的操控话术识别与反制系统属于 AI 内容安全 / 文本分类工具核心技术关键词规则基线 开源大模型语义分类 FastAPI 接口封装推荐硬件CPU 可以跑小参数量化模型GPU 推理体验更好显存占用需按实际模型版本测试支持平台Windows / Linux / macOS 均可依赖 Python 环境启动方式命令行启动模型服务FastAPI 提供 Web 服务可扩展 WebUI是否支持 API支持提供文本检测、批量检测、健康检查三类接口是否支持批量任务支持可读取 CSV 批量识别带日志和失败重试机制主要功能话术风险分类、操控标签提取、反制建议生成、批量识别、结果导出适合场景情感安全工具、社群内容审核、客服对话质检、聊天助手防骗模式这套方案的定位不是做一个“完美裁判”而是做一个“低门槛、能跑通、可扩展”的识别框架。先跑通主流程再根据实际数据和业务需求调整模型和 Prompt这才是更稳妥的做法。2. 方案设计为什么用 AI 识别操控话术2.1 传统关键词过滤的局限如果只做关键词过滤例如把“你不配”“除了我没人在乎你”“你离不开我”这类句子加入黑名单实现起来很快但效果很粗糙。操控话术本质上是一种“语义模式”而不是固定文本组合同一个意图可以换一百种说法。比如“你离开我肯定没人要”和“你看看谁会要你”表达的核心都是打压但关键词层面很难泛化。更麻烦的是关键词名单会产生大量误报普通朋友之间的玩笑话也可能被误伤。2.2 大模型的语义理解优势大模型擅长做的事情恰恰是“从上下文里理解意图”。给模型一段对话让它判断是否存在打压、贬低、制造焦虑、情感勒索、煤气灯效应等操控特征它会结合上下文、语气、角色关系综合输出分类结果。相比关键词系统它更像一个“初级审查员”而不是一把只能切固定位置的刀。2.3 系统整体架构从工程实现角度看整套系统分成五层数据层保存合规获取的对话样本脱敏后按标签整理成 JSON 或 CSV。规则层先用关键词或正则做粗筛命中后再送大模型精判降低 API 压力和成本。模型层通过 Ollama 或 vLLM 加载开源大模型本地推理数据不出内网。服务层FastAPI 提供检测接口支持单条和批量两种模式。应用层Web 页面、命令行工具、CSV 批量导入导出。这里规则层的作用很实际先用低成本的规则过滤明显正常文本只把疑似内容交给大模型这样在批量场景下能省不少算力。从架构角度说这个设计还有一个隐藏价值规则层可以不断从模型层的学习结果里沉淀关键词两个模块互相迭代系统会越来越准。第一次跑通时可以先不做这个循环但目录结构上建议给规则库单独留文件。3. 环境准备与前置条件3.1 软件和硬件清单这是一个偏轻量的文本分类项目硬件门槛远低于图像和视频生成。如果只是测试CPU 跑 7B 或更小的量化模型也能出结果只是速度慢一些如果要做批量任务建议有独立显卡。下面是一套通用的环境检查清单具体版本以实际安装为准检查项说明操作系统Windows 10/11、Ubuntu 20.04、macOS 均可Python建议 3.10 或更高版本模型运行时Ollama 或 vLLM二选一CPU 场景优先 OllamaWeb 框架FastAPI Uvicorn依赖库requests、pandas、pydanticGPU 驱动NVIDIA 用户建议安装最新驱动和 CUDA 工具包磁盘空间模型文件通常几 GB 到十几 GB按实际下载模型大小预留3.2 安装 Python 依赖# 创建虚拟环境Windows 和 Linux 命令一致路径按实际项目调整 python -m venv .venv # 激活虚拟环境 # Windows .venv\Scripts\activate # Linux / macOS source .venv/bin/activate # 安装依赖 pip install fastapi uvicorn requests pandas pydantic3.3 安装模型运行时以 Ollama 为例安装完成后先确认服务能正常启动# 查看 Ollama 版本 ollama --version # 拉取一个开源中文对话模型这里以 qwen 系列为例具体版本名以官方仓库为准 ollama pull qwen2.5:7b如果你本机已经通过 FastChat、llama.cpp 或 vLLM 部署过模型也可以直接复用。Ollama 只是本教程选用的“最省事”方案并不是唯一方案。启动后不要急着进下一步先跑一句测试确认模型能正常返回内容。3.4 数据准备和合规提醒整个项目最关键的数据来源是“合规获取的对话样本”。可以是自己标注的模拟对话、公开数据集、经过授权脱敏的客服聊天记录也可以是你在自己账号权限范围内采集的公开评论。不管来源是哪里都要遵守三条底线涉及个人信息的文本必须脱敏去掉姓名、手机号、账号等标识信息。涉及真实聊天记录的需要获得相关方授权或者干脆只用于本地实验不对外发布。项目输出只能用于反操控、内容安全、合规审查等正当用途不得用于指导实施任何操控行为。4. 构造话术样本库与规则基线4.1 样本标签体系要让大模型输出稳定结果先得定义一套清晰的标签体系。建议从下面几类开始不要一开始就搞很细标签含义示例特征打压贬低否定对方价值削弱自信“你这条件没人要的”制造焦虑制造紧迫感和危机感“你现在不改变以后就完了”情感勒索用“付出-回报”绑架对方“我为你付出那么多你要听我的”煤气灯效应让对方怀疑自己的记忆和判断“是你记错了我根本没说过”正常沟通无明显操控特征“今天工作怎么样”注意这部分写的是“识别维度”不是“操作方法”。标签体系的价值在于让模型有统一的输出格式避免同一个意思被模型用各种不同的词汇表达出来。4.2 样本库格式建议使用 JSON 保存标注样本每一条包含编号、文本、标签和备注。格式如下[ { id: 1, text: 除了我没人会这么容忍你了。, label: 打压贬低, note: 典型打压句式 }, { id: 2, text: 你现在不按我说的做以后有你后悔的。, label: 制造焦虑, note: 制造紧迫感 }, { id: 3, text: 我不是那个意思是你太敏感了。, label: 煤气灯效应, note: 转移责任并否定对方感受 } ]这个样本库有两个用途一是用来测试模型分类效果二是以后如果要微调模型这就是现成的训练数据底子。建议一开始每个标签准备 20 到 50 条样本先看效果再决定要不要扩大。4.3 规则基线规则层不建议写太复杂先放 10 到 20 个高频词即可。代码里用一个列表保存后续可以从模型的结果里反推新词再补充RULE_KEYWORDS { 打压贬低: [没人要, 你不行, 将就, 除了我, 受不了你], 制造焦虑: [错过就没了, 不改变就完了, 你迟早会后悔], 情感勒索: [我为你付出, 你要感恩, 必须听我的], 煤气灯效应: [你太敏感, 你记错了, 我没说过, 你想多了], } def rule_check(text: str): hit_rules [] for label, keywords in RULE_KEYWORDS.items(): for kw in keywords: if kw in text: hit_rules.append(label) break return hit_rules规则命中后把文本送进大模型精判规则没命中的可以直接标记为“正常沟通”也可以抽样送模型复查。如果规则命中率和模型结论差异很大就要回头调整关键词表。5. 本地部署大模型与调用5.1 启动模型服务Ollama 安装完成后默认会监听本地端口。先手动确认模型能跑ollama run qwen2.5:7b看到模型可以正常对话后退出交互模式进入 API 调用阶段。Ollama 默认的 API 地址一般为http://127.0.0.1:11434/api/generate实际以官方文档为准。下面用 curl 简单验证curl http://127.0.0.1:11434/api/generate \ -H Content-Type: application/json \ -d { model: qwen2.5:7b, prompt: 用一句话分析除了我没人会这么容忍你了。, stream: false }如果返回结果里包含正常文本模型服务就可以进入下一步了。这里特别提醒model参数必须替换为你本机实际拉取的模型名不同版本的模型名可能不一样。5.2 构建分类 Prompt让模型做分类Prompt 要尽可能给几个明确约束输出格式固定、标签范围固定、不要解释过度。一个比较稳定的 Prompt 模板如下你是一个沟通话术安全审查助手。请判断用户输入的内容是否包含操控式沟通话术包括但不限于打压贬低、制造焦虑、情感勒索、煤气灯效应。 只输出 JSON格式如下 {label: 唯一标签, reason: 简短理由, suggestion: 反制建议} 用户输入{text}suggestion字段很关键它让这个系统不只是“识别危险”还能给出反制方向。比如识别出“打压贬低”时反制建议可以是“提醒对方可以用建设性方式表达并重申边界”。这是整套系统最有实用价值的部分也是它区别于普通敏感词系统的地方。5.3 用 Python 调用模型import requests import json OLLAMA_URL http://127.0.0.1:11434/api/generate def classify_text(text: str, model: str qwen2.5:7b): prompt f你是一个沟通话术安全审查助手。请判断用户输入的内容是否包含操控式沟通话术包括但不限于打压贬低、制造焦虑、情感勒索、煤气灯效应。 只输出 JSON格式如下 {{label: 唯一标签, reason: 简短理由, suggestion: 反制建议}} 用户输入{text} payload { model: model, prompt: prompt, stream: False } response requests.post(OLLAMA_URL, jsonpayload, timeout60) result response.json() content result.get(response, ).strip() # 实际使用中需要加 JSON 解析容错处理 return content if __name__ __main__: print(classify_text(除了我没人会这么容忍你了。))这一步跑通之后整个系统的“识别大脑”就完成了。接下来要做的是把分类能力包成一个人人能访问的 API 服务。6. 构建话术识别 API 服务6.1 创建 FastAPI 应用from fastapi import FastAPI, HTTPException from pydantic import BaseModel import requests app FastAPI(titleConnection Safety API) OLLAMA_URL http://127.0.0.1:11434/api/generate MODEL_NAME qwen2.5:7b class TextRequest(BaseModel): text: str class BatchRequest(BaseModel): texts: list[str] def build_prompt(text: str) - str: return f你是沟通话术安全审查助手。请判断用户输入是否包含操控式沟通话术包括打压贬低、制造焦虑、情感勒索、煤气灯效应。 只输出 JSON{{label: 唯一标签, reason: 简短理由, suggestion: 反制建议}} 用户输入{text} app.get(/health) def health(): return {status: ok} app.post(/detect) def detect(req: TextRequest): prompt build_prompt(req.text) try: resp requests.post(OLLAMA_URL, json{ model: MODEL_NAME, prompt: prompt, stream: False }, timeout60) result resp.json().get(response, ) return {input: req.text, result: result} except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/batch_detect) def batch_detect(req: BatchRequest): results [] for text in req.texts: try: prompt build_prompt(text) resp requests.post(OLLAMA_URL, json{ model: MODEL_NAME, prompt: prompt, stream: False }, timeout60) result resp.json().get(response, ) results.append({input: text, result: result}) except Exception as e: results.append({input: text, error: str(e)}) return {results: results}保存为api_server.py然后启动uvicorn api_server:app --host 127.0.0.1 --port 8000启动后访问http://127.0.0.1:8000/health返回{status:ok}就说明服务正常。6.2 接口调用示例curl -X POST http://127.0.0.1:8000/detect \ -H Content-Type: application/json \ -d {text: 你离开我肯定没人要}Python 调用示例import requests url http://127.0.0.1:8000/detect payload {text: 你离开我肯定没人要} response requests.post(url, jsonpayload, timeout60) print(response.json())6.3 批量任务设计批处理不要简单用 for 循环硬跑建议做三件事输入输出分开目录避免污染原数据。每处理一条写一行日志方便断点续跑。失败请求自动重试三次重试间隔递增。下面是一个简单的 CSV 批处理示例import csv import time import requests API_URL http://127.0.0.1:8000/detect def run_batch(input_csv, output_csv): with open(input_csv, encodingutf-8) as fin, \ open(output_csv, w, encodingutf-8, newline) as fout: reader csv.DictReader(fin) writer csv.writer(fout) writer.writerow([text, label, reason, suggestion]) for row in reader: text row[text] for attempt in range(3): try: resp requests.post(API_URL, json{text: text}, timeout60) result resp.json().get(result, ) writer.writerow([text, result]) break except Exception as e: print(f第 {attempt 1} 次失败: {text}, 错误: {e}) time.sleep(2 * (attempt 1)) if __name__ __main__: run_batch(input.csv, output.csv)7. 功能测试与效果验证7.1 单条检测测试准备几条不同风格的输入分别调用/detect接口看模型输出是否与预期接近测试输入预期方向判断标准“今天天气不错一起去散步吧”正常沟通label 不落入风险标签“你离开我肯定没人要”打压贬低label 包含打压特征“你现在不按我说的做以后肯定后悔”制造焦虑label 包含焦虑特征“是你记错了我从来没说过”煤气灯效应label 包含否定事实特征不要要求模型 100% 准确重点是“高风险文本不能放过低风险文本不能大量误报”。如果明显正常文本频繁被标记为风险说明 Prompt 或模型选择需要调整。7.2 批量测试用 100 条左右混合样本跑一遍批量接口统计两类问题漏报率明明是高风险话术但模型判为正常。误报率普通沟通被模型判为高风险。第一次跑建议小批量先看结果再扩大数据量。真正上线前建议拿 500 条以上经过标注的测试集做一次正式评估记录各项指标。7.3 判断是否成功的标准满足以下三条就可以认为这套小系统“可用”所有高风险标签都能被映射到预设标签体系之一。正常沟通文本大部分不会被误判为风险。suggestion反制建议有参考价值而不是空话套话。7.4 常见失败原因如果模型输出经常是无效 JSON或者标签超出预设范围优先检查 Prompt 是否明确限制了输出格式。其次是模型本身指令遵循能力不足可以换更大或更新的模型。最后才是调整温度参数把温度降低到 0 到 0.3 之间输出会更稳定。8. 性能与资源占用观察8.1 怎么观察占用模型推理过程中的资源占用要在模型加载后、推理过程中观察而不是看空闲状态。GPU 用户命令行执行nvidia-smi观察显存和 GPU 利用率CPU 用户打开任务管理器或top命令观察 CPU 和内存。8.2 CPU 和 GPU 的差异同一个模型在 CPU 上推理耗时通常远高于 GPU尤其是 7B 以上模型。如果只是偶尔测几条CPU 可以接受如果要做批量任务建议准备一张独立显卡并优先选择量化版本模型降低显存占用。具体数字不用在意重点是根据自己的设备先跑一个最小测试记录单条耗时再推算批量处理时间。8.3 降低资源占用的方法优先使用量化模型例如 Q4 或 Q5 版本。限制模型最大生成长度没必要生成大段文字。批量任务控制并发数不要一次性把几十条请求全部打给模型。规则层提前过滤明显正常文本减少无谓推理。9. 常见问题排查问题现象可能原因排查方式解决方案Ollama 无法拉取模型网络问题或模型名错误查看错误日志、确认模型名更换下载源或按官方仓库确认模型名模型响应很慢CPU 推理或模型过大查看 CPU/GPU 占用换小模型或量化版控制并发GPU 不识别驱动或 CUDA 未正确安装运行nvidia-smi检查更新驱动检查环境变量显存不足模型太大或并发过高查看显存占用换量化模型限制并发API 返回超时单条推理时间过长查看后端日志增加 timeout减少并发模型输出非 JSONPrompt 约束不足查看原始响应内容强化 Prompt设置更低温度批量任务中途中断单条异常未捕获查看日志定位加 try/except、断点续跑、失败重试误报率过高样本量不足或模型不匹配抽样分析补充样本调 Prompt调整规则词10. 最佳实践与合规边界这套系统的价值在于“识别和防护”而不是“教授操控”。所以在实际部署时要特别注意边界只部署在受控环境中接口服务默认绑定127.0.0.1不要直接暴露到公网。需要远程访问时应增加认证鉴权。输入文本必须脱敏真实聊天记录不能直接进样本库。涉及人脸、声音、个人信息等敏感数据的场景要通过授权和隐私合规审核。系统输出的“反制建议”仅供参考重大风险场景仍需要人工介入。不要根据模型识别结果对真实用户进行公开定性避免误伤。如果需要对外展示重点展示技术流程和检测能力不要展示他人对话原文。工程实践方面建议保留一套最小可运行配置一个小模型、一份少量样本、一个 FastAPI 服务入口。后续所有尝试都基于这套基线展开改动前先保存可回滚版本。模型文件、输入素材、输出结果分目录管理日志单独存放。批量任务要限制并发和超时时间加上失败重试日志。11. 总结与下一步这个项目最值得尝试的地方不是拿一个大模型跑分类而是亲手把一个模糊的“沟通风险”概念变成了可量化的工程系统。你用规则层用大模型做语义判断用 FastAPI 接成接口再用 CSV 批量跑完一轮真实样本整个链路就闭合了。第一步先验证什么很简单把第 4 节样本库里的内容跑一遍看模型能不能稳定输出 JSON、是否能覆盖五个预设标签。这个验证不花多少时间却决定了后续所有方向能不能往下走。最容易踩的坑有两个一是语料不足导致模型误报率高看起来效果很差其实是样本和 Prompt 的问题不是模型不行二是批量任务不做超时控制一旦模型响应变慢整个任务就卡死。提前做好重试和日志会省很多事。后续能扩展的方向不少。可以把规则层和模型层做成相互迭代的闭环每隔一段时间从模型结果中抽取新关键词回填规则库可以接入聊天机器人在对方发送高风险话术时即时提醒可以做成 Web 管理台把识别记录、统计报表、人工复核整合在一个界面里也可以把识别结果导出成结构化数据用于更细粒度的沟通质量分析。这套方案的方向是用大模型把“隐藏的操控模式”变成“可见的标签和报告”。建议收藏备用后面做内容安全、社群风控、聊天助手相关的功能时可以直接把中间的分类服务和批量识别思路拿出来复用。