基于Coze搭建智能客服系统：对话记录与购买意向分析实战-平芜编程栈

背景痛点：客服对话这座“金矿”为何挖不动？

做电商的朋友都懂，每天几万条客服聊天里藏着大量“我要买”“再便宜点”这类金句。可现实是：

聊天记录散落在各个客服客户端，格式千奇百怪，Excel 打开就卡死；
运营同学人肉筛对话，一天看不了几百条，还常常漏掉高意向用户；
老板临时要“本周高意向用户列表”，技术团队只能通宵跑脚本，第二天还被吐槽“结果不准”。

一句话：非结构化数据没法直接用，人工分析又慢又废眼。于是我们把目光投向了智能客服 + 自动意向分析，让机器先帮我们把“金矿”筛出来。

技术选型：为什么最后选了 Coze？

在动手之前，我们给主流方案做了张打分表（10 分制，越高越好）：

维度	Rasa	Dialogflow	Coze
中文语料原生支持	6	7	9
可视化调试界面	5	8	9
私有化部署成本	7（开源）	3（按量计费）	8（容器镜像）
对话分析插件生态	6	6	9（官方自带内容理解模块）
二次开发灵活度	9	6	8（Python SDK 友好）

Rasa 自由度最高，但中文 NLU 组件要自己训，时间耗不起；Dialogflow 汉化一般，且流量一大钱包就瘪；Coze 在“中文理解 + 可视化 + 可私有”三点上最均衡，于是拍板：就它了。

核心实现：让客服机器人边聊边记账

1. 整体架构

先放一张总览图，后面按模块拆。

2. 对话记录存储：MongoDB 分片集群

设计思路

对话流持续追加，写远大于读，选 MongoDB；
按 tenant+date 分片，避免热片；
单条文档结构（精简版）：

{ "_id": "conv_20250625_001", "tenant": "shop_a", "ts": ISODate("2025-06-25T10:00:00Z"), "msgs": [ {"from": "user", "text": "这款有 128G 吗？", "ts": ...}, {"from": "bot", "text": "有的，现货", "ts": ...} ], "intents": ["ask_stock", "price_sensitive"], "purchase_score": 0.82 }

关键代码（Python 3.11）

# storage.py import os import logging from datetime import datetime from pymongo import MongoClient, errors from pymongo.write_concern import WriteConcern logger = logging.getLogger("coze.storage") MONGO_URI = os.getenv("MONGO_URI", "mongodb://mongos-router:27017") DB_NAME = "coze_conv" COLL_NAME = "dialogue" wc_majority = WriteConcern("majority", wtimeout=5000) class ConvStorage: def __init__(self): self.client = MongoClient(MONGO_URI, uuidRepresentation="standard") self.coll = self.client[DB_NAME][COLL_NAME] def insert_conv(self, conv_id: str, tenant: str, msgs: list): doc = { "_id": conv_id, "tenant": tenant, "ts": datetime.utcnow(), "msgs": msgs, "intents": [], "purchase_score": -1, } try: self.coll.insert_one(doc, write_concern=wc_majority) logger.info("conv %s saved", conv_id) except errors.DuplicateKeyError: logger.warning("conv %s already exists", conv_id)

3. 购买意向分析：BERT 微调 + 规则引擎

标注阶段

先让运营同学用 Coze 内置的“内容标注”插件，把 3 千条历史对话打上“高 / 中 / 低”意向标签；
导出 JSONL，字段：text、label。

微调脚本

# train_intent.py from datasets import load_dataset from transformers import BertForSequenceClassification, Trainer, TrainingArguments dataset = load_dataset("json", data_files="intent.jsonl") model = BertForSequenceClassification.from_pretrained("bert-base-chinese", num_labels=3) args = TrainingArguments( output_dir="./intent_model", per_device_train_batch_size=32, num_train_epochs=3, logging_steps=50, evaluation_strategy="epoch", save_strategy="epoch", metric_for_best_model="eval_accuracy", ) trainer = Trainer(model=model, args=args, train_dataset=dataset["train"], eval_dataset=dataset["test"]) trainer.train() trainer.save_model("./intent_model")

推理封装（带缓存 & 日志）

# intent_predictor.py import torch, logging, os from transformers import BertTokenizerFast, BertForSequenceClassification from functools import lru_cache logger = logging.getLogger("coze.intent") class IntentPredictor: def __init__(self, model_dir: str = "./intent_model"): self.tokenizer = BertTokenizerFast.from_pretrained("bert-base-chinese") self.model = BertForSequenceClassification.from_pretrained(model_dir) self.model.eval() self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") self.model.to(self.device) @torch.no_grad() def predict(self, text: str) -> int: inputs = self.tokenizer(text, return_tensors="pt", truncation=True, padding=True, max_length=128) inputs = {k: v.to(self.device) for k, v in inputs.items()} logits = self.model(**inputs).logits label = int(logits.argmax(-1).item()) logger.debug("text=%s -> label=%s", text, label) return label

规则引擎兜底

BERT 再强也怕“领域外”句子，于是加两条正则：

价格类：r"便宜|优惠|券|减|折"
库存类：r"现货|发|多久到"

若正则命中且 BERT 给出“低”意向，自动升一档，减少漏杀。

4. 把分析结果写回 MongoDB

# updater.py from storage import ConvStorage from intent_predictor import IntentPredictor import re price_re = re.compile(r"便宜|优惠|券|减|折") stock_re = re.compile(r"现货|发|多久到") def calc_purchase_score(intent_label: int, text: str) -> float: score_map = {0: 0.9, 1: 0.6, 2: 0.2} # 高/中/低 base = score_map.get(intent_label, 0.1) if price_re.search(text) or stock_re.search(text): base = min(base + 0.15, 0.95) return base def update_conv(conv_id: str, msgs: list): storage = ConvStorage() predictor = IntentPredictor() intents = [] total = 0.0 for m in msgs: if m["from"] == "user": label = predictor.predict(m["text"]) intents.append(label) total += calc_purchase_score(label, m["text"]) avg_score = total / max(len(intents), 1) storage.coll.update_one( {"_id": conv_id}, {"$set": {"intents": intents, "purchase_score": round(avg_score, 2)}} )

性能优化：高并发也能扛住

1. 对话异步处理

客服消息实时性要求 <200 ms，但意向分析可接受秒级延迟。于是把“更新对话”任务丢进队列（用 Redis Stream）：

# async_worker.py import redis, json, logging, os from updater import update_conv r = redis.Redis(host=os.getenv("REDIS_HOST", "redis"), decode_responses=True) stream_key = "coze_conv_stream" group = "analyzer" consumer = "worker-1" try: r.xgroup_create(stream_key, group, id="0", mkstream=True) except redis.ResponseError: pass # 已存在 def main(): while True: msgs = r.xreadgroup(group, consumer, {stream_key: ">"}, count=10, block=1000) for _, items in msgs: for _id, fields in items: try: update_conv(fields["conv_id"], json.loads(fields["msgs"])) r.xack(stream_key, group, _id) except Exception as e: logging.exception("process failed: %s", fields)

2. 模型热加载

模型文件 400 M，重启一次耗时 5-6 s，对灰度不友好。用 symlinks + inotify：

新模型放到 intent_model-v2 目录；
训练脚本执行完校验后，ln -sfn intent_model-v2 intent_model；
predictor 进程监听inotify.Modify，发现 symlink 变化即重新from_pretrained，老请求继续用旧模型，新请求进新模型，实现零中断切换。

避坑指南：别等踩坑才 Google

1. 对话数据脱敏

姓名、手机、地址用正则先遮罩，例如1****2345；
训练集导出前跑一遍pii_scan，把身份证、银行卡号整行替换为<PII>，防止模型“背下来”。

2. 意图模型冷启动

上线第一天没标注数据？用“弱标签”方案：

把历史成交用户对话关键词当正样本（“下单”“付款”），负样本随机抽；
先用 TF-IDF + 逻辑回归训个 baseline，让运营边用边标，两周后替换 BERT 微调模型，准确率从 0.68 提到 0.87。

3. 高并发下消息丢失

Redis Stream 开xadd时指定MAXLEN ~100000，防止内存爆；
客服端发消息时把conv_id 写入本地 SQLite，若 5 s 内未收到 ACK 则重推，保证“至少一次”。

生产建议：监控先行，别等老板拍桌子

指标	采集方式	告警阈值
意图识别准确率	每日随机 500 条人工复核	<85% 飞书 + 电话
响应延迟 P99	Prometheus Summary	>500 ms
MongoDB 写延迟	db.serverStatus()	>200 ms
队列堆积长度	redis xlen	>5000