企业官网建设流程全解析

问题背景

“您好，请稍等，正在为您转接人工客服……”
这句熟悉的开场白，平均要让人等 30-60 秒。传统客服系统大多基于“关键词+正则”或“if/else 规则树”，维护成本高、扩展性差，一旦业务上新，就要重新写规则、重新发版。更尴尬的是，用户换种说法问同一问题，机器人就“装傻”，体验瞬间翻车。

过去三年，我们团队先后踩过三种技术路线：

1. 规则引擎：开发一周即可上线，但半年后规则膨胀到 3 万条，维护人员从 2 人变成 8 人，依旧挡不住“问法稍变就失效”。
2. 传统 NLP（意图分类+槽位填充）：用 BERT 微调，意图识别准确率 88%，看着还行，可一旦遇到多轮追问，上下文就“断片”，用户得重复提供信息。
3. 大模型：参数大到 7B/13B，自带“常识”，无需穷举规则，端到端就能做语义理解与生成，准确率直接拉到 95%+，同时支持多轮、多语言、情绪识别。

对比之后，我们决定把宝押在大模型上，但很快发现“会跑 demo”和“敢上生产”是两回事：并发一高 GPU 就爆显存、问答一深就丢上下文、用户一调皮就触发敏感词。于是有了下面这份“从 0 到 1”的实战笔记。

架构设计

整体思路一句话：“异步解耦 + 缓存提速 + 弹性扩缩”。先上图，再拆讲。

1. 接入层

• Nginx 统一入口，按 location 拆分“普通问答”和“敏感问答”双集群，方便做蓝绿发布。
• 网关集成 Sentinel，做 QPS 限流与熔断。

2. 业务层

• 采用“生产者-消费者”模型，FastAPI 接收请求后只负责校验参数→写消息到 Redis Stream→立即返回 202，前端通过 SSE 长连接轮询结果，体感“秒回”。
• 消费者（Worker）按需横向扩容，K8s HPA 以 CPU>60% 或队列积压>500 为指标自动弹升。

3. 模型层

• 选型：经内部 2000 条业务语料评测，13B 参数模型在准确率、吞吐、成本三点平衡最好；最终采用 Llama-2-13B-Chat，量化到 INT4，单卡 A10G 可跑。
• 推理框架：vLLM 连续批处理（continuous batching），相比 HuggingFace generate 提升 3.2× 吞吐。
• 缓存：Redis 缓存“问题→向量签名”的 TTL=300s，重复问法直接命中，减少 28% GPU 调用。

4. 数据层

• 对话状态存储在 MongoDB，按 session_id 分片，索引覆盖“user_id+timestamp”。
• 向量检索用 Milvus，存储 FAQ 向量库，Top5 召回后作为 Prompt 上下文，降低幻觉。

核心实现

下面给出最常被问的三块代码：负载均衡调用、对话状态管理、敏感词过滤。全部单文件可跑，已按 PEP8 格式化，可直接集成到现有仓库。

1. 异步推理客户端（带缓存+负载均衡）

# llm_client.py import asyncio import hashlib import httpx from typing import List from redis import asyncio as aioredis CACHE_TTL = 300 WORKERS: List[str] = ["http://gpu-worker-1:8000", "http://gpu-worker-2:8000"] class LlmClient: def __init__(self): self.rds = aioredis.from_url("redis://redis:6379/0", decode_responses=True) self.client = httpx.AsyncClient(limits=httpx.Limits(max_connections=100)) async def ask(self, user_id: str, query: str) -> str: # 1. 构造缓存 key key = "cache:" + hashlib.md5(query.encode()).hexdigest() cached = await self.rds.get(key) if cached: return cached # 2. 轮询负载均衡选 worker worker = WORKERS[hash(user_id) % len(WORKERS)] payload = {"prompt": query, "max_tokens": 512, "temperature": 0.3} resp = await self.client.post(f"{worker}/generate", json=payload, timeout=15) resp.raise_for_status() answer = resp.json()["text"] # 3. 写缓存并返回 await self.rds.setex(key, answer, ex=CACHE_TTL) return answer

2. 对话状态管理（支持多轮、指代消解）

# dialog_state.py from datetime import datetime from typing import Dict, List from pymongo import MongoClient, ASCENDING, IndexModel mongo = MongoClient("mongodb://mongo:27017/chat") session_coll = mongo["chat"]["session"] # 建复合索引，防止全表扫描 session_coll.create_indexes([IndexModel([("user_id", ASCENDING), ("ts", ASCENDING)])]) class DialogState: """负责拉取、追加、压缩多轮上下文""" MAX_HISTORY = 6 # 经验值，再大显存吃不消 @staticmethod async def load(user_id: str) -> List[Dict[str, str]]: cursor = session_coll.find({"user_id": user_id}).sort("ts", -1).limit(MAX_HISTORY) return [{"role": x["role"], "content": x["content"]} for x in cursor][::-1] @staticmethod async def append(user_id: str, role: str, content: str): session_coll.insert_one({ "user_id": user_id, "role": role, "content": content, "ts": datetime.utcnow() })

3. 敏感词过滤（DFA+拼音混淆+回车断句）

# sensitive_filter.py import re from typing import Set class DFAFilter: def __init__(self, word_set: Set[str]): self.root = {} for w in word_set: self._add_word(w.lower()) def _add_word(self, word: str): node = self.root for ch in word: node = node.setdefault(ch, {}) node["end"] = True def exists(self, text: str) -> bool: text = re.sub(r"\s+", "", text.lower()) for i in range(len(text)): node = self.root for ch in text[i:]: if ch not in node: break node = node[ch] if "end" in node: return True return False

把上面三段代码拼进 Worker，主循环大概长这样：

async def consumer(): while True: msg = await redis.xreadgroup({"stream": "chat", "group": "gpt", "consumer": "c1"}) user_id, query = parse_msg(msg) history = await DialogState.load(user_id) answer = await llm.ask(user_id, build_prompt(history, query)) await DialogState.append(user_id, "assistant", answer) await redis.xack("chat", "gpt", msg.id)

性能调优

1. QPS 测试方法

• 工具：locust + custom client，模拟 1k 并发，SSE 长连接。
• 指标：①首包时间（TTFT）<800 ms；②系统 QPS≥120；③P99 延迟<2 s。
• 结果：单卡 A10G + vLLM 连续批，峰值 QPS 140，TTFT 中位数 320 ms，符合业务 SLA。

2. 显存优化技巧

• 采用 4-bit 量化，显存占用从 26 GB 降到 13 GB；再开 ZeRO-3 offload，把优化器状态放内存，再省 3 GB。
• 设置--gpu-memory-utilization 0.9，预留 10% 给临时 Tensor，防止 OOM。
• 对话长度超过 1k token 时，后台触发 Summary 线程，把历史压缩成 100 字“摘要”，再写回 Mongo，减少下次输入。

3. 超时熔断策略

• 推理侧：单请求>20 s 未返回，vLLM 自动 cancel，返回 504；网关层 Sentinel 统计 504 比例>5% 即熔断该节点，流量漂到另一节点。
• 缓存侧：Redis 单节点故障，Client 快速失败（retry_on_timeout=False），本地内存 LRU 兜底 30 s，保证核心链路可用。

生产实践

1. 意图识别漂移

• 现象：大模型对“开发票”与“开票信息”两个意图打错标签，导致下游接口调错。
• 解决：①在 Prompt 里加“Few-Shot”，给 3 组正例 2 组反例；②每周拉最新 500 条误分类数据，人工标注→LoRA 微调 1 epoch，漂移率从 8% 降到 2%。

2. 多轮对话上下文丢失

• 现象：用户问“那手续费呢？”，模型忘了前文谈的是“退票”。
• 解决：把系统角色消息里注入“领域标记”，如“【领域:退票】”，并在 Summary 里保留关键实体，测试显示指代消解准确率提升 18%。

3. 敏感词过滤误杀

• 现象：“我操，这票价真贵”被误判含脏话，直接拒绝。
• 解决：DFA 白名单 + 情绪模型二次确认；白名单里放“操作为”“操心”等 120 个常用组合，误杀率从 1.2% 降到 0.15%。

4. 上线 checklist（血泪版）

• GPU 节点一定加nvidia-fabricmanager，否则驱动升级后 P2P 通信会掉速 30%。
• 推理服务端口别用 8000，和 K8s healthz 冲突，改 8080 省得半夜 oncall。
• 发版前跑一轮 2h 压力测试，显存占用是否匀速上涨，若持续上涨=有内存泄漏，立即 rollback。

延伸思考

把系统怼到 140 QPS 后，新的矛盾出现了：
“再拉大模型尺寸到 30B，效果还能涨 2 个点，可 TTFT 直奔 1.2 s，客服主管不答应；砍回 7B，延迟是稳了，准确率又掉 3 个点，运营不答应。”

如何优雅地平衡“效果 vs 延迟”？目前能想到三条路：

1. 动态路由：简单 FAQ 走 7B，复杂投诉走 30B，用轻量分类器做调度。
2. 投机解码（Speculative Decoding）：小模型打草稿，大模型做批改，理论 2× 提速。
3. 边缘+云混合：边缘 7B 保证首响，云侧 30B 异步修正答案，再推送给前端做“无痕刷新”。

哪条路更香？或者你有第四条路？欢迎留言一起拆坑。