营销投放跑不动问题往往不在渠道预算也不在文案不够花哨而在于你没法用一句客户自己的话说清楚他在为什么付钱。说不出痛点后面所有的关键词投放、内容选题、销售话术都变成蒙眼开枪子弹打出去靠的是猜。这篇不聊空泛的营销理论聊一套能落地的技术流程——把“客户痛点”当成一个可以被采集、被抽取、被量化、被排序的数据对象来处理。具体做法是四段式流水线先把客服工单、售后评论、社群问答、丢单记录这些原始文本统一采集并脱敏再做清洗与主题聚类把散落的抱怨压成若干候选簇然后用大模型做结构化抽取让每条痛点带上场景、触发条件、现有替代方案、损失代价和原始证据最后算出一个可排序的痛点优先级并直接生成能拿去投放和销售复述的表达句式。门槛并不高。聚类和统计部分在普通 CPU 机器上就能跑完真正吃资源的是大模型抽取那一步而这一步既可以走本地推理也可以走兼容 OpenAI 协议的接口服务具体显存占用和吞吐要按你实际选的模型版本和批量大小测试本文不给死数字。接口和批量任务部分我会给出可以直接复制的 FastAPI 服务和批处理队列模板方便接到自己的工单系统里。如果你在做 B 端产品定位、SaaS 增长、独立开发选题或者单纯想让销售团队别再背那些客户听不懂的话术下面这套流程可以照着搭一遍。1. 核心能力速览先把这套东西的规格摆清楚省得看到一半才发现跑不起来。它不是一个开箱即用的商业软件而是一条可以自己拼装的分析链路每一段都能单独替换。能力项说明落地方式输入数据类型客服工单、售后评价、社群问答、丢单原因记录、站内搜索词统一转成 CSV 或 JSONL字段名对齐处理链路清洗脱敏 → 主题聚类 → 结构化抽取 → 优先级排序 → 表达生成Python 脚本串联可分段执行硬件门槛统计与聚类在 CPU 上可跑结构化抽取按实际模型版本测试需按实际环境测试显存占用材料未提供具体数值取决于所选模型与批量大小以本机实测为准启动方式脚本单次运行或 FastAPI 常驻服务uvicorn 启动接口能力HTTP POST输入一批文本返回结构化痛点 JSON通用模板路径按项目调整批量能力目录级批处理带失败重试与断点续跑队列文件 状态标记输出格式JSONL 明细 汇总表格便于二次分析直接进 BI 或表格工具适合场景产品定位、内容选题、销售话术、竞品差评分析需配合人工复核不适合场景无授权的数据采集、伪造评价、骚扰式外呼明确排除这张表里最需要强调的一点是“需按实际环境测试”。很多人在本地跑通一次小样本就以为整条链路通了结果一上量就卡在内存和超时上。正确做法是先固定一套最小可运行配置一百条左右样本、单进程、批量大小为一跑通全链路之后再逐项放大。2. 痛点判定标准与常见误区在写任何代码之前得先定义清楚什么叫做“说得出来的痛点”。否则模型抽出来一堆听起来很像那么回事的短句你依然没法判断哪条能用。我的判断标准是四个要素齐全缺一个都只能算抱怨不能算痛点。第一个要素是场景具体。必须是“月底对账的时候导出三张表手工核对”这种能还原出动作的描述而不是“效率低”“体验差”这种放之四海皆准的形容词。第二个要素是损失可量化。时间损失、金钱损失、机会损失至少要能落到一个数量级上哪怕是“每次多花半小时”也比“很麻烦”有用。第三个要素是现有替代方案明确并且这个方案有摩擦。客户不是活在真空里的他今天一定在用某个办法凑合你的机会就藏在这个凑合方案的摩擦点里。第四个要素是有原话证据。每条痛点后面必须挂一段客户自己说的原话方便销售复述也方便内容团队直接改成标题。对照这四条几种常见的“假痛点”就很好识别了下面这张表是内部评审时最常用的对照清单。常见表述问题所在改成什么客户希望提升效率无场景、无损失财务每周手工合并 5 张对账表平均 3 小时客户觉得价格贵未区分“预算不足”与“价值感知不足”单次采购超 2 万需走三层审批周期拖到 6 周客户说功能不全未锁定具体缺口缺少按项目维度的成本拆分只能导总账客户反馈不好用无触发条件新人在第 3 天批量导入时找不到字段映射入口客户说再考虑考虑是决策链问题不是产品问题采购需技术负责人与财务双签技术侧已通过这个表建议直接做成评审卡每次从模型输出里挑候选痛点逐条过一遍。过不了的打回重抽不要硬塞进营销素材里。很多团队的问题不是没有数据而是把抱怨当成了痛点最后写出来的文案客户一看就知道你没做过这行。3. 适用场景、数据源与合规边界这套流程适合三类人。第一类是做 B 端产品的市场同学手里有一堆工单和丢单记录但没时间逐条看第二类是独立开发者或小团队想在做之前先确认某个需求到底痛不痛第三类是内容和服务商需要从竞品差评里找选题和差异点。这三类的共同点是手上有真实文本缺的是结构化的整理能力。不适合的场景也要说清楚。如果你手上没有真实反馈数据只有自己的猜测那这套流程跑出来的东西只是把猜测包装得更像数据反而更危险。如果数据来源涉及未授权的抓取、绕过平台限制、采集他人隐私内容那就不要做这不是技术问题而是合规问题。凡是涉及用户评价、客服对话、通话记录这类内容必须做到三件事采集有授权或属于平台公开可用的合规范围、存储前完成脱敏、使用范围限定在内部分析与产品改进不外传、不二次售卖、不用于骚扰式触达。涉及的手机号、身份证号、详细地址、订单号等字段在进入分析流水线之前就要替换掉下面的清洗脚本会处理这一层。另外提醒一句痛点分析的产出是“决策参考”不是“事实结论”。模型抽出来的频次高低只代表在你这份语料里的分布不代表市场全貌。语料有偏结论就有偏这一点在样本量小于几百条时尤其明显。4. 环境准备与四段式流水线环境这块不需要多复杂Python 一个虚拟环境加上几个常规库就够了。大模型那一步如果是调接口本地几乎不占资源如果是本地推理按照模型官方给出的显存要求准备即可务必先小批量验证再放大。下面是基础依赖的安装命令Windows 和 Linux 都适用第二行激活命令按系统选一条。python -m venv venv # Linux / macOS source venv/bin/activate # Windows PowerShell # venv\Scripts\Activate.ps1 pip install pandas jieba scikit-learn fastapi uvicorn requests tqdm目录结构建议一开始就分清楚输入、中间结果、输出、日志各占一个目录后面批量跑的时候不容易乱。这个习惯在小样本阶段看不出价值一旦数据量上去、需要断点续跑就会感谢自己当初分了目录。painpoint/ ├── data/ │ ├── raw/ # 原始导出只读不做任何修改 │ └── clean/ # 清洗脱敏后的 JSONL ├── work/ │ ├── clusters/ # 聚类中间结果 │ └── extracted/ # 模型抽取结果按批次分文件 ├── out/ │ ├── painpoints.jsonl │ └── matrix.csv ├── logs/ └── scripts/四段式流水线对应四个脚本clean.py负责清洗脱敏cluster.py负责分词聚类出候选簇extract.py负责调模型做结构化抽取score.py负责算优先级并导出矩阵。每段都读写文件不搞内存里一把梭这样任何一段出问题都能单独重跑不用从头再来。5. 数据清洗与主题聚类实现清洗这一步的目标不是把文本洗得多干净而是把噪声控制在不影响聚类的程度。常见的坑有三个一是同一用户的重复提交没去重导致某个抱怨的频次被虚高二是短文本太多五六个字的“不好用”“退款”会把聚类中心带偏三是客服模板话术混在里面比如“您好已为您登记”这些属于流程语言不是客户语言。所以在长度过滤之外还要加一层模板句过滤。下面这段脚本做了四件事读取原始 CSV、按文本去重、做基础脱敏、过滤过短文本和模板句输出统一格式的 JSONL。字段名需要按你实际的导出格式调整这里用的是通用示例。import re import json import pandas as pd # 基础脱敏规则按实际数据补充 PATTERNS { phone: re.compile(r1[3-9]\d{9}), idcard: re.compile(r\d{17}[\dXx]), email: re.compile(r[\w.\-][\w\-]\.[\w.]), } TEMPLATE_HINTS [已为您登记, 请您耐心等待, 感谢您的反馈, 稍后将与您联系] def desensitize(text: str) - str: for key, pattern in PATTERNS.items(): text pattern.sub(f{key}, text) return text def is_noise(text: str) - bool: if len(text) 12: return True return any(hint in text for hint in TEMPLATE_HINTS) def clean(input_csv: str, output_jsonl: str) - None: df pd.read_csv(input_csv) df df.dropna(subset[content]) df[content] df[content].astype(str).map(desensitize) df df.drop_duplicates(subset[content]) df df[~df[content].map(is_noise)] with open(output_jsonl, w, encodingutf-8) as f: for _, row in df.iterrows(): record { id: str(row.get(id, )), source: str(row.get(source, unknown)), content: row[content], } f.write(json.dumps(record, ensure_asciiFalse) \n) print(f清洗完成保留 {len(df)} 条) if __name__ __main__: clean(data/raw/tickets.csv, data/clean/tickets.jsonl)清洗完之后做聚类。这一步的目的不是让机器告诉你痛点是什么而是把几百上千条文本压成十几个候选簇让人工能在一小时内看完。TF-IDF 加 KMeans 是最省事的组合不需要 GPU几百条数据秒级出结果。需要注意的是中文要先分词并且把“的、了、是、我们”这类高频虚词加进停用词表否则聚类中心会被这些词占满。import json import jieba from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.cluster import KMeans STOPWORDS set([的, 了, 是, 我们, 你们, 他们, 这个, 那个, 然后, 但是]) def load_texts(path: str): texts, ids [], [] with open(path, encodingutf-8) as f: for line in f: item json.loads(line) texts.append(item[content]) ids.append(item[id]) return texts, ids def tokenize(text: str) - str: return .join(w for w in jieba.cut(text) if w not in STOPWORDS and len(w) 1) def run_cluster(path: str, k: int 12): texts, ids load_texts(path) corpus [tokenize(t) for t in texts] vectorizer TfidfVectorizer(max_features3000, ngram_range(1, 2)) matrix vectorizer.fit_transform(corpus) model KMeans(n_clustersk, random_state42, n_init10) labels model.fit_predict(matrix) terms vectorizer.get_feature_names_out() for idx in range(k): center model.cluster_centers_[idx] top center.argsort()[-8:][::-1] keywords [terms[i] for i in top] size int((labels idx).sum()) print(f簇 {idx} 样本数 {size}{ / .join(keywords)}) return ids, labels if __name__ __main__: run_cluster(data/clean/tickets.jsonl, k12)簇的数量 k 建议按样本量估一百条左右用 8 到 12上千条可以用 20 以上再人工合并。输出的关键词列表就是候选痛点方向的线索比如某簇里反复出现“对账、合并、导出、表格”那基本可以确定是财务对账这条线。这时候不要急着写文案把该簇的原话抽十条出来读一遍你对痛点的理解会比看关键词准得多。6. 大模型结构化痛点抽取聚类给的是方向结构化抽取给的是能进表格的结论。这一步的关键不在提示词写得多花哨而在于强制模型输出固定字段不允许自由发挥。字段设计我建议就按前面说的痛点四要素来再补两个量化字段方便后面排序。下面这份 JSON 就是要求的输出结构字段名可以按你的下游系统调整但类型要固定。{ pain_point: 一句话概括痛点不超过30字, scene: 发生在什么角色、什么时间点、什么动作之前, trigger: 什么条件下这个问题会被触发, current_workaround: 客户现在用什么办法凑合摩擦点在哪, cost: 损失描述尽量带数量级, cost_type: time | money | risk | labor, evidence: 客户原话摘录必须来自输入文本, severity: 1, confidence: 0.0 }severity用 1 到 5 的整数让模型自己打分会比较主观但比没有强后面还要靠人工校准。confidence是模型对这条抽取的把握程度低于某个阈值的一律进人工复核队列。最关键的是evidence字段必须约束成“只能摘录原文不许改写”这样你才能一眼看出模型是不是在编。调用部分给一个兼容 OpenAI 协议格式的通用模板。本地推理服务、自建网关、第三方接口只要是这个协议都能直接换地址用不需要改逻辑。请求路径和端口按你的实际服务替换批量大小先设成一跑通再放大。import json import time import requests API_URL http://127.0.0.1:8000/v1/chat/completions # 按实际服务替换 MODEL_NAME your-model-name # 按实际部署的模型替换 SYSTEM_PROMPT 你是用户研究分析师。根据给定文本抽取客户痛点严格输出 JSON。 要求 1. evidence 必须是原文摘录不得改写、不得编造 2. 文本中没有明确场景时scene 填 unknown 3. 一条文本最多输出一个痛点无有效痛点时返回 {pain_point: none} 4. 只输出 JSON不要输出解释。 def extract_one(text: str, retries: int 3) - dict: payload { model: MODEL_NAME, messages: [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: text[:2000]}, ], temperature: 0.2, } for attempt in range(retries): try: resp requests.post(API_URL, jsonpayload, timeout120) resp.raise_for_status() content resp.json()[choices][0][message][content] return json.loads(content) except (requests.RequestException, json.JSONDecodeError, KeyError) as exc: print(f第 {attempt 1} 次失败{exc}) time.sleep(2 ** attempt) return {pain_point: error} if __name__ __main__: result extract_one(每个月月底都要把三张表导出来手工核对经常加班到十点) print(json.dumps(result, ensure_asciiFalse, indent2))这段代码里有三个细节值得单独说。第一是超时设成 120 秒本地推理在小批量下经常比接口慢超时太短会大量误判为失败。第二是重试用了指数退避二次重试间隔 2 秒、4 秒避免服务刚启动还没预热就被打满。第三是异常捕获里包含了json.JSONDecodeError因为模型偶尔会在 JSON 前后加一句“好的结果如下”这种情况要当成失败重试而不是解析失败直接崩掉。抽取跑完之后先别急着信结果。抽十条出来把evidence拿去和原文比一遍看看有没有改写、有没有张冠李戴。这一步只花五分钟但能省掉后面大量的返工。如果发现模型倾向于把所有文本都归纳成“效率低”那多半是提示词里缺少反例加一句“禁止使用效率低、体验差等空泛表述”通常就能改善。7. 痛点矩阵排序与营销表达生成有了结构化的痛点清单下一步是排序。排序公式不用搞太复杂能体现业务判断就够了。我常用的是频次、严重度、付费信号三个维度相乘再除以解决成本。付费信号是个代理指标可以用“提到预算、审批、采购、续费”这类词的出现次数近似也可以用该条文本所属客户的实际成交状态加权后者更准但需要 CRM 数据打通。import json import pandas as pd def load_extracted(path: str) - pd.DataFrame: rows [] with open(path, encodingutf-8) as f: for line in f: item json.loads(line) if item.get(pain_point) in (none, error): continue rows.append(item) return pd.DataFrame(rows) def build_matrix(df: pd.DataFrame) - pd.DataFrame: df[pain_point] df[pain_point].str.strip() agg df.groupby(pain_point).agg( freq(pain_point, size), severity(severity, mean), confidence(confidence, mean), cost_type(cost_type, lambda s: s.mode().iloc[0] if len(s) else unknown), ).reset_index() # 解决成本为人工预估1 表示容易3 表示很难 agg[solve_cost] 1.0 agg[priority] (agg[freq] * agg[severity] * agg[confidence]) / agg[solve_cost] return agg.sort_values(priority, ascendingFalse) if __name__ __main__: df load_extracted(work/extracted/batch_01.jsonl) matrix build_matrix(df) matrix.to_csv(out/matrix.csv, indexFalse, encodingutf-8-sig) print(matrix.head(10).to_string(indexFalse))排序出来之后按四象限做取舍高频高严重的是主打卖点高频低严重的是基础体验项低频高严重的是细分场景的机会低频低严重的可以先放着。这个分层直接决定了营销资源的分配主打卖点拿去投放基础体验项拿去优化产品细分机会拿去写长尾内容。表达生成这一步不要交给模型自由发挥。给它一个固定句式让它填坑效果会稳定很多。我常用的模板是“当……的时候你不得不……结果是……”三个空分别对应 scene、current_workaround、cost。生成完之后做一个很土的测试把这句话念给一线销售听问他能不能用自己的话复述出来并且举一个客户例子。如果他说不出来这句话就是不合格的跟文案好不好看没关系。再补一个验证手段把生成的表达拿去和站内搜索词、投放关键词做匹配。如果客户搜索时用的词和你的表达用词完全对不上说明你在用自己的语言体系说话转化率不会好。这个匹配可以直接用现成的搜索词报告做不需要额外成本。8. 接口化、批量任务与性能观察单次跑脚本适合探索阶段一旦要接进工单系统或者按周自动更新就得做成服务。下面这个 FastAPI 服务只做一件事接收一批文本返回结构化痛点列表。生产环境记得加上鉴权、限流和请求体大小限制不要直接裸暴露在公网。from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titlePainPoint Extractor) class ExtractRequest(BaseModel): texts: list[str] max_items: int 50 class ExtractResponse(BaseModel): items: list[dict] app.post(/api/v1/painpoints, response_modelExtractResponse) def extract(req: ExtractRequest) - ExtractResponse: texts req.texts[: req.max_items] results [extract_one(t) for t in texts] return ExtractResponse(itemsresults) app.get(/healthz) def health() - dict: return {status: ok}启动和调用都很直接端口按实际空闲端口调整启动前先确认没被占用。uvicorn app:app --host 127.0.0.1 --port 8010 --workers 1curl -X POST http://127.0.0.1:8010/api/v1/painpoints \ -H Content-Type: application/json \ -d {texts:[月底对账要手工合并五张表经常加班],max_items:10}批量任务建议用配置文件驱动而不是把参数写死在脚本里。下面这份配置就是最小可用的批处理描述输入目录、输出目录、批量大小、并发数、失败重试次数都在里面跑批的时候只需要改文件不用改代码。{ input_dir: ./data/clean, output_dir: ./work/extracted, batch_size: 20, concurrency: 1, retry: 3, resume: true }concurrency一开始务必设成 1。很多人图快设成 8结果本地推理服务直接被压垮全是超时最后排查半天以为是模型问题。等单并发稳定跑完上千条再往上加到 2 或 4同时观察显存和响应时间。resume打开之后已经输出过的记录会跳过中断了直接重跑即可。性能观察主要看三个指标单条平均耗时、失败率、显存或内存占用曲线。单条耗时可以用日志里的时间戳差值算失败率看重试次数和最终error的比例占用曲线用系统自带的资源监视工具看就行重点看有没有随批量增大而持续上涨的趋势——如果有多半是某处缓存没清。降低占用的手段按性价比排序是减小批量、缩短输入文本长度超长文本先切句再抽、降低并发、换更小的模型。分辨率、步数这类参数在这条链路里不适用因为处理的是文本不是图像别把图像生成的调参思路套上来。9. 常见问题排查与最佳实践实际跑下来出问题的位置高度集中下面这张表覆盖了大部分情况。遇到问题先按表查比自己从日志第一行读起快得多。问题现象可能原因排查方式解决方案依赖安装失败Python 版本过低或缺少编译工具链看报错里是否有版本要求提示升级到 3.10 以上改用预编译包分词结果全是单字停用词表为空或文本含大量符号打印前 20 个 token 观察补停用词表先做一次符号清洗聚类簇大小极不均衡k 值过大或存在大量重复文本看各簇样本数分布先去重再适当减小 k模型返回不是合法 JSON提示词约束不足或输出被截断打印原始返回内容强制只输出 JSON适当提高最大输出长度抽取结果 evidence 与原文不符模型改写原话抽样逐条比对提示词强调原文摘录校验失败即重试接口调用频繁超时并发过高或服务未预热看服务端日志和资源占用降并发到 1加长超时先跑小批量批量任务中途卡住单条请求无超时或进程残留查端口占用与进程列表给请求加超时重启服务并开启断点续跑输出痛点全是空泛词语料本身信息量不足抽十条原文人工读换数据源或补充访谈记录不要硬凑显存或内存持续上涨缓存未清理结果全堆在内存里观察占用曲线改成流式写文件分批释放最佳实践这块给几条成本最低的。第一第一次永远用小样本一百条足够暴露 90% 的问题别一上来就全量。第二把data/raw设为只读任何清洗都在副本上做原始数据永远留一份。第三模型抽出的每条痛点都必须挂evidence没有原话支撑的一律不进营销素材这条规则坚持下来能挡掉大部分自嗨内容。第四人工复核不要省模型负责压缩信息人负责判断真假两者的分工不能混。第五涉及用户文本、通话记录、评价内容的数据脱敏在入口做不要等到输出再处理分析结果只在内部使用不对外披露具体客户信息。如果这条链路你已经跑顺了接下来可以往两个方向扩。一是把痛点矩阵和实际投放数据打通看哪些痛点对应的素材转化更好反过来修正排序权重二是把抽取结果做成知识库销售在跟进时能按客户行业、角色检索对应的痛点原话。这两个方向的投入都比继续调提示词更值。回到标题那句话说不出客户的痛点营销就是蒙眼开枪。而“说得出来”这件事本质上是个信息处理问题你有没有把散落在工单、评价、聊天记录里的真实语言收集起来有没有把它们压成可比较的结构有没有用原话去验证你的判断。这套流程不保证你的文案一定爆但它至少能保证你打出去的每一枪瞄准的是一个真实存在的靶子。建议先把最小可运行版本跑通从你自己手上最熟悉的那批客户反馈开始跑完一轮再决定要不要上量。