基于大模型的营销智能客服实现：从架构设计到生产环境部署实战-编程实验室

背景痛点：传统规则引擎在营销场景下的三大瓶颈

去年“618”大促，我们组的客服系统被用户吐槽到飞起：

促销话术更新滞后：运营凌晨改价格，规则库要到第二天中午才上线，结果早上 9 点用户问“前 100 名半价还有吗？”，机器人只会回“请稍等，转人工”。
用户画像利用不足：老客复购意向模型已经算到 87%，对话里却还在推新客券，转化率惨不忍睹。
多轮对话维护难：活动页叠券逻辑一改，开发要在 2000 条 if-else 里找分支，上线两周后自己都看不懂。

一句话：规则引擎在营销这种“高频、高并发、高变化”的三高场景下，基本跑不动。

技术选型：GPT-4 vs Claude 3 实测对比

我们拿 5 万条真实营销语料做了离线评测，指标三件套：意图准确率、情感 F1、平均延迟。

模型	意图准确率	情感 F1	平均延迟	商用成本/1k session
GPT-4	94.2%	0.91	680 ms	0.42$
Claude 3 Sonnet	92.7%	0.89	410 ms	0.30$
自研 7B 微调	88.4%	0.85	190 ms	0.08$

结论：

延迟敏感 → Claude 3 胜出，410 ms 在 500 TPS 下 GPU 利用率还能压到 65%。
商用成本 → 自研 7B 最低，但准确率差 6 个点，营销场景不能忍。
最终拍板：线上用 Claude 3，兜底用自研 7B，做级联降级。

核心实现：LangChain + RAG + 异步流水线

1. 对话状态机：把“营销套路”画成图

营销客服最大的套路就是“发券→领券→下单→追单”。用 LangChain 的GraphState十分钟就能画出来：

from langchain.graph import StateGraph, END class MarketingState: START = "START" COUPON_PUSH = "COUPON_PUSH" ORDER_ASK = "ORDER_ASK" CHASE_ORDER = "CHASE_ORDER" END = "END" transitions = [ (MarketingState.START, "ask_coupon", MarketingState.COUPON_PUSH), (MarketingState.COUPON_PUSH,"ask_order", MarketingState.ORDER_ASK), (MarketingState.ORDER_ASK, "no_order", MarketingState.CHASE_ORDER), (MarketingState.CHASE_ORDER,"finish", MarketingState.END), ] graph = StateGraph(states=MarketingState, transitions=transitions)

状态节点里再塞 RAG 提示词，就能把“券”讲清楚。

2. 带缓存的向量检索：别让 Milvus 被 QPS 打爆

营销知识库 30 万条 SKU+活动，纯向量检索 QPS 上 500 直接跪。我们加了一层 Redis 缓存，key 是“用户问题 3-gram 指纹”，value 是 top-5 结果 ID 列表，TTL 300 s。

import hashlib from typing import List import redis, pymilvus class CachedRetriever: def __init__(self, milvus_collection, redis_client): self.col = milvus_collection self.rds = redis_client def search(self, text: str, top_k: int = 5) -> List[str]: key = hashlib.md5(text.encode()).hexdigest() cached = self.rds.get(key) if cached: return cached.decode().split(",") # 走 Milvus vec = self._encode(text) res = self.col.search(vec, top_k) ids = [r.id for r in res[0]] self.rds.setex(key, 300, ",".join(ids)) return ids

3. 异步流水线：把 CPU 等 GPU 的时间省出来

Claude 3 一次生成 600 ms，IO 等待占 70%。我们用 FastAPI + asyncio + aioproducer，把“向量检索→LLM 调用→后处理”拆成三阶段队列：

网关层收到请求即返回“已受理”，把事件推入 Kafka。
消费端异步做检索+LLM，结果写回 Redis Stream。
网关层 SSE 推送给前端轮询延迟从 800 ms 降到 220 ms。

压测结果：单卡 A10 从 120 QPS 提到 420 QPS，GPU 利用率 92%。

生产考量：压测、日志、脱敏

1. JMeter 配置要点

线程组：500 并发，Ramp-up 60 s，循环 300 次。
报文：JSON，带 Authorization Bearer，动态提取 user_id 用 CSV 数据集。
断言：响应码 200 + JSON 字段"status=="answered"。
监控：Backend Listener 打到 InfluxDB，Grafana 看 P99 延迟曲线。

2. 对话日志脱敏

正则一把梭：

import re def desensitize(text: str) -> str: text = re.sub(r"\d{11}", "<PHONE>", text) text = re.sub(r"\d{17}[\dXx]", "<ID>", text) return text

落库前再调用，字段级 AES-加密，密钥放 KMS，审计表只存哈希。

避坑指南：幻觉、版本、热更新

1. 大模型幻觉三件套

双重检索校验：RAG 返回的 top-1 相似度 < 0.82 时，强制走“暂无答案”模板。
规则后置过滤：正则黑名单 247 条，含“100% 治愈”“绝对赚钱”等广告法雷区。
置信度打分：让模型输出[[confidence:0.87]]，后端阈值 0.75，以下走人工兜底。

2. 知识库增量更新版本控制

Milvus 里给每个 doc 加version_tag，格式yyyymmddhhmm。上线流程：

新数据写带新 version 的集合。
灰度 5% 流量，对比点击率/转人工率。
全量切换后，旧集合延迟 24 h 删除，支持秒级回滚。

代码规范：PEP8 + 防御式编程

关键函数示例：

from typing import Optional import logging def query_llm(prompt: str, max_tokens: int = 512) -> Optional[str]: if not isinstance(prompt, str) or not prompt.strip(): raise ValueError("prompt must be non-empty string") if not (0 < max_tokens <= 4096): raise ValueError("max_tokens out of range") try: resp = claude_client.completions.create(prompt=promptmax_tokens=max_tokens) return resp.text except Exception as e: logging.exception("llm call failed") return None