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

有些事放到技术语境里看，会变得特别有意思。一个教别人用话术操控关系的“大师”，他总结的那套“打压-拉扯-制造焦虑-推拉配合”的操作流程，本质上是一套可以被穷举的文本模式。以前识别这种模式要靠人肉判断，一条一条看聊天记录、看社群发言、看直播话术，效率极低。但现在不一样了，这套模式喂给大模型，再用 Prompt 约束输出，就能做到批量识别、打标签、生成反制建议，甚至直接封装成 API 接进审核系统。

这篇文章就把这套“反向工具”完整拆一遍：怎么用自己的聊天记录或公开合规语料构造样本库，怎么用本地部署的开源大模型做语义分类，怎么用 FastAPI 把识别能力封装成接口，以及怎么跑批量任务、观察显存占用、排查常见问题。这里说的“PUA 识别”，指的是识别操控式沟通话术，用于情感安全、内容审核、合规质检等方向，不是教人怎么实施话术。整套方案的核心，是把模糊的“感觉被拿捏”变成可量化的标签，让套路本身变成可被程序识别的数据。

适合看这篇文章的读者主要有三类：第一类是 NLP 或 AI 应用工程师，想找一个真实场景练手，把大模型接进业务系统；第二类是社群运营、内容安全、客服质检相关岗位，需要自动筛选高风险对话；第三类是产品经理或独立开发者，想快速搭一个 AI 工具 Demo，验证“AI 内容安全”方向的产品价值。下面直接进正文，先看整体能力设计，再一步步从零搭建。

1. 核心能力速览

这套方案的定位不是做一个“完美裁判”，而是做一个“低门槛、能跑通、可扩展”的识别框架。先跑通主流程，再根据实际数据和业务需求调整模型和 Prompt，这才是更稳妥的做法。

2. 方案设计：为什么用 AI 识别操控话术

2.1 传统关键词过滤的局限

如果只做关键词过滤，例如把“你不配”“除了我没人在乎你”“你离不开我”这类句子加入黑名单，实现起来很快，但效果很粗糙。操控话术本质上是一种“语义模式”而不是固定文本组合，同一个意图可以换一百种说法。比如“你离开我肯定没人要”和“你看看谁会要你”表达的核心都是打压，但关键词层面很难泛化。更麻烦的是，关键词名单会产生大量误报，普通朋友之间的玩笑话也可能被误伤。

2.2 大模型的语义理解优势

大模型擅长做的事情恰恰是“从上下文里理解意图”。给模型一段对话，让它判断是否存在打压、贬低、制造焦虑、情感勒索、煤气灯效应等操控特征，它会结合上下文、语气、角色关系综合输出分类结果。相比关键词系统，它更像一个“初级审查员”，而不是一把只能切固定位置的刀。

2.3 系统整体架构

从工程实现角度看，整套系统分成五层：

• 数据层：保存合规获取的对话样本，脱敏后按标签整理成 JSON 或 CSV。
• 规则层：先用关键词或正则做粗筛，命中后再送大模型精判，降低 API 压力和成本。
• 模型层：通过 Ollama 或 vLLM 加载开源大模型，本地推理，数据不出内网。
• 服务层：FastAPI 提供检测接口，支持单条和批量两种模式。
• 应用层：Web 页面、命令行工具、CSV 批量导入导出。

这里规则层的作用很实际：先用低成本的规则过滤明显正常文本，只把疑似内容交给大模型，这样在批量场景下能省不少算力。

从架构角度说，这个设计还有一个隐藏价值：规则层可以不断从模型层的学习结果里沉淀关键词，两个模块互相迭代，系统会越来越准。第一次跑通时可以先不做这个循环，但目录结构上建议给规则库单独留文件。

3. 环境准备与前置条件

3.1 软件和硬件清单

这是一个偏轻量的文本分类项目，硬件门槛远低于图像和视频生成。如果只是测试，CPU 跑 7B 或更小的量化模型也能出结果，只是速度慢一些；如果要做批量任务，建议有独立显卡。下面是一套通用的环境检查清单，具体版本以实际安装为准：

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 pydantic

3.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, json=payload, timeout=60) 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(title="Connection 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 }, timeout=60) result = resp.json().get("response", "") return {"input": req.text, "result": result} except Exception as e: raise HTTPException(status_code=500, detail=str(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 }, timeout=60) 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, json=payload, timeout=60) 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, encoding="utf-8") as fin, \ open(output_csv, "w", encoding="utf-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}, timeout=60) 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接口，看模型输出是否与预期接近：

不要要求模型 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. 常见问题排查

10. 最佳实践与合规边界

这套系统的价值在于“识别和防护”，而不是“教授操控”。所以在实际部署时要特别注意边界：

• 只部署在受控环境中，接口服务默认绑定127.0.0.1，不要直接暴露到公网。需要远程访问时，应增加认证鉴权。
• 输入文本必须脱敏，真实聊天记录不能直接进样本库。
• 涉及人脸、声音、个人信息等敏感数据的场景，要通过授权和隐私合规审核。
• 系统输出的“反制建议”仅供参考，重大风险场景仍需要人工介入。
• 不要根据模型识别结果对真实用户进行公开定性，避免误伤。
• 如果需要对外展示，重点展示技术流程和检测能力，不要展示他人对话原文。

工程实践方面，建议保留一套最小可运行配置：一个小模型、一份少量样本、一个 FastAPI 服务入口。后续所有尝试都基于这套基线展开，改动前先保存可回滚版本。模型文件、输入素材、输出结果分目录管理，日志单独存放。批量任务要限制并发和超时时间，加上失败重试日志。

11. 总结与下一步

这个项目最值得尝试的地方，不是拿一个大模型跑分类，而是亲手把一个模糊的“沟通风险”概念变成了可量化的工程系统。你用规则层，用大模型做语义判断，用 FastAPI 接成接口，再用 CSV 批量跑完一轮真实样本，整个链路就闭合了。

第一步先验证什么？很简单，把第 4 节样本库里的内容跑一遍，看模型能不能稳定输出 JSON、是否能覆盖五个预设标签。这个验证不花多少时间，却决定了后续所有方向能不能往下走。

最容易踩的坑有两个：一是语料不足导致模型误报率高，看起来效果很差，其实是样本和 Prompt 的问题，不是模型不行；二是批量任务不做超时控制，一旦模型响应变慢，整个任务就卡死。提前做好重试和日志，会省很多事。

后续能扩展的方向不少。可以把规则层和模型层做成相互迭代的闭环，每隔一段时间从模型结果中抽取新关键词回填规则库；可以接入聊天机器人，在对方发送高风险话术时即时提醒；可以做成 Web 管理台，把识别记录、统计报表、人工复核整合在一个界面里；也可以把识别结果导出成结构化数据，用于更细粒度的沟通质量分析。

这套方案的方向，是用大模型把“隐藏的操控模式”变成“可见的标签和报告”。建议收藏备用，后面做内容安全、社群风控、聊天助手相关的功能时，可以直接把中间的分类服务和批量识别思路拿出来复用。