背景痛点：传统客服的“三座大山”
做客服系统最怕三件事：
- 用户问一句，后台转三圈，5 秒后才蹦出“正在输入”；
- 同一句话换种说法，机器人就“失忆”，答非所问；
- 618 大促流量一冲，MySQL 直接打挂，人工坐席全线爆炸。
归根结底，老系统把“意图识别 + 多轮状态 + 知识检索”三件事揉在一个同步接口里，一旦模型体积变大或 QPS 升高，冷启动、上下文丢失、扩展性差的问题就集体爆发。
技术选型：为什么选了 DeepSeek + Python
| 维度 | DeepSeek | 某 7B 开源模型 | 云端大厂 API |
|---|---|---|---|
| 中文表现 | 自带 2T 中文语料，法律/电商场景识别准 | 需额外 30% 中文语料微调 | 够用，但按 Token 计费 |
| 推理成本 | 量化后 4-bit，单卡 A10 可压 2000 RPM | 8-bit 需 24G 显存 | 按量计费，大促账单惊人 |
| 私有化 | 可完全离线，满足合规 | 同左 | 数据出域审批难 |
| Python 生态 | 官方提供 transformers 插件，LangChain 已适配 | 社区自己搓轮子 | 仅 REST，SDK 黑盒 |
结论：DeepSeek 在“效果/成本/可控”三角里相对均衡，再叠加 Python 的 FastAPI + LangChain + Redis 全家桶，开发效率最高。
核心实现：三步搭出最小可用骨架
1. 意图识别：让 DeepSeek 听懂人话
# intent_server.py from fastapi import FastAPI from transformers import AutoTokenizer, AutoModelForCausalLM import torch, json, os MODEL_PATH = "/models/deepseek-7b-chat" tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH, trust_remote_code=True) model = AutoModelForCausalLM.from_pretrained( MODEL_PATH, torch_dtype=torch.float16, device_map="auto" ) app = FastAPI() def build_prompt(user_query: str, hist: list) -> str: """ 构造带历史的多轮 prompt，强制模型只输出 JSON， 避免啰嗦前置语，提高下游解析成功率。 """ hist_text = "\n".join([f"User:{q}\nBot:{a}" for q, a in hist]) prompt = ( "你是客服机器人，请根据对话历史判断用户最新问题的意图，" "输出合法 JSON 且仅有 intent 字段。\n" f"{hist_text}\nUser:{user_query}\n" "Intent:" ) return prompt @app.post("/intent") async def get_intent(query: str, session_id: str): from redis_handler import get_hist # 见下节 hist = get_hist(session_id) prompt = build_prompt(query, hist) inputs = tokenizer(prompt, return_tensors="pt").to(model.device) with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=20, temperature=0.3, # 调低随机，保证意图稳定 do_sample=True, ) answer = tokenizer.decode(outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True) try: return json.loads(answer.strip()) except Exception: return {"intent": "unknown"}要点：
- 强制 JSON 输出，下游直接
json.loads，省掉正则； - Temperature 降到 0.3，分类场景不需要创造性；
- 历史对话拼进 prompt，单轮也能利用多轮信息。
2. 对话状态管理：Redis 五秒落地
# redis_handler.py import redis, json, os, time r = redis.Redis(host=os.getenv("REDIS_HOST", "localhost"), decode_responses=True) def get_hist(sid: str, turn=5) -> list: data = r.lrange(f"chat:{sid}", -turn, -1) return [json.loads(x) for x in data] def add_turn(sid: str, query: str, answer: str, ttl=3600): r.lpush(f"chat:{sid}", json.dumps({"q": query, "a": answer}, ensure_ascii=False)) r.expire(f"chat:{sid}", ttl)用 List 结构天然保序，lrange 取最近 N 轮，ttl 自动清掉僵尸会话，省内存。
3. RAG 增强：向量库 + 重排两步走
# rag_retriever.py from langchain.vectorstores import FAISS from langchain.embeddings import HuggingFaceBgeEmbeddings from langchain.schema import Document import numpy as np EMBED_PATH = "BAAI/bge-base-zh" embeddings = HuggingFaceBgeEmbeddings(model_name=EMBED_PATH) texts = [ "7 天无理由退货：需保证商品完好，附件齐全。", "运费险理赔：签收 72h 内申请，按实际运费赔。", ... ] docs = [Document(page_chunk) for chunk in texts] vector_db = FAISS.from_documents(docs, embeddings) def retrieve(query: str, top_k=5): return vector_db.similarity_search(query, k=top_k)召回后，再用 Cross-Encoder 重排，把最相关 3 条塞进 prompt，实测比纯向量提升 18% 答案准确率。
生产环境：让 QPS 从 50 飙到 380
1. 异步处理 + 协程池
FastAPI 原生支持async/await，但模型推理是 CPU 密集型，直接放协程里会阻塞事件循环。
解决：开 4 进程gunicorn -k uvicorn.workers.UvicornWorker --workers 4，再配torch.set_num_threads(1)，把每个 worker 的 CPU 线程钉死，避免争抢。
2. 流式返回，降低首 Token 延迟
from transformers import TextIterator @app.post("/stream") async def stream_answer(query: str): prompt = build_full_prompt(query) inputs = tokenizer(prompt, return_tensors="pt").to(model.device) streamer = TextIterator(tokenizer, skip_prompt=True) generation_kwargs = dict( inputs, streamer=streamer, max_new_tokens=512, temperature=0.7 ) # 非阻塞生产 import threading threading.Thread(target=model.generate, kwargs=generation_kwargs).start() for new_text in streamer: yield f"data: {json.dumps({'token': new_text})}\n\n"前端 Event 收包即可实时渲染，用户感知延迟从 2.1 s 降到 0.6 s。
3. 隐私保护：对话历史加密落盘
- 敏感字段（手机号、地址）用
presid==>占位符替换后再存； - Redis 开启
ACL + TLS，磁盘持久化关闭，只当缓存； - 每日凌晨跑
redis --bigkeys巡检，超 24h 的 key 强制淘汰。
4. 负载测试数据（单节点 4*A10）
- QPS：峰值 380，平稳 250
- P99 延迟：intent 接口 220 ms，RAG 问答 1.8 s
- GPU 显存占用：7.3 GB / 24 GB
避坑指南：踩过的坑，能救一个是一个
微调过拟合：
准备 5 万条客服日志，按 8:1:1 切，训练 3 epoch 后指标狂掉。后来加early_stopping_patience=1+dropout=0.1，测试集才稳住。对话流超时：
网关默认 30 s 断连接，而模型偶尔 35 s 才吐。网关层加proxy_read_timeout 60s;同时业务侧抛asyncio.TimeoutError给前端，让它自动重试/reconnect接口，带原 session_id 续聊。敏感词过滤：
开源敏感库只覆盖广告法，电商场景不够。把自家历史工单 2000 个脏词做成 Trie 树，配合flashtext0.2 ms 级替换，再让 DeepSeek 二次校验，召回率 96%，误杀率 <1%。
结尾：开放讨论
大模型效果越好，成本越高。我们目前用 4-bit 量化 + 流式 + 协程，把单卡榨到 250 QPS，但大促高峰仍要弹性扩容。
你所在团队是如何平衡“模型容量”与“响应速度”的？欢迎聊聊你们的剪枝、蒸馏或动态路由方案。
