背景痛点：智能客服到底难在哪？
做 ToB SaaS 这些年，我最大的感受是——客服场景是“看起来简单，做起来全是坑”。多轮对话只是冰山一角，真正的暗礁在下面三层：
- 状态维护：用户中途改需求、跳话题、甚至回退到三句之前，模型必须记得住、跟得上。
- 领域知识：企业 FAQ 往往散落在 Confluence、PDF、旧工单里，格式不统一，更新频率高，模型答错一次就“社死”。
- 延迟敏感：网页端>800 ms 的响应，用户就开始多点刷新；语音通道更苛刻，>300 ms 就明显卡顿。
- 合规红线：聊天记录含手机号、订单号、地址，一旦泄露，罚单比算力贵得多。
以上四点，决定了“选个大模型 API 直接怼上去”必然翻车。下面按实战时间线，记录我们团队从 0 到 1 的爬坑笔记。
技术选型：GPT、Claude 还是本地化小模型？
我们把“成本/效果/速度”拆成 9 个可量化指标，跑了两周压测，结论先给：
| 维度 | gpt-3.5-turbo | claude-3-sonnet | Llama3-8B（本地） | |---|---|---|---|---| | 首 token 延迟（P95） | 450 ms | 380 ms | 120 ms | | 每 1K session 成本 | 0.42 $ | 0.55 $ | 0.03 $（电费） | | 微调门槛 | 仅 RLHF、收费 | 仅 Prompt | 全量+LoRA 可玩 | | 中文 SFT 后 BLEU↑ | — | — | +18% | | 数据出境风险 | 有 | 有 | 无 | | 并发 200 下 GPU 占用 | — | — | 单卡 A100 占 75% |
最终 hybrid 方案：
- 80% 常规咨询 → 本地 Llama3-8B + LoRA，延迟<200 ms，成本忽略不计。
- 20% 复杂工单 → 丢给 Claude，牺牲 100 ms 换质量，预算可控。
核心实现一：对话状态机（带缓存+超时）
大模型无状态，我们就给它“外挂”一个状态机。Python 3.11 代码可直接贴生产，带类型标注与异常捕获。
from __future__ import annotations import time import json from dataclasses import dataclass, asdict from threading import Lock from typing import Dict, Optional @dataclass class Turn: role: str content: str timestamp: float class DialogueState: """线程安全会话级缓存""" def __init__(self, ttl_seconds: int = 600): self._store: Dict[str, list[Turn]] = {} self._ttl = ttl_seconds self._lock = Lock() def _is_expired(self, session_id: str) -> bool: turns = self._store.get(session_id, []) return turns and (time.time() - turns[-1].timestamp) > self._ttl def append(self, session_id: str, turn: Turn) -> None: with self._lock: if self._is_expired(session_id): self._store.pop(session_id, None) self._store.setdefault(session_id, []).append(turn) def get_history(self, session_id: str, k: int = 10) -> list[dict]: with self._lock: if self._is_expired(session_id): self._store.pop(session_id, None) return [] turns = self._store.get(session_id, [])[-k:] return [asdict(t) for t in turns] state = DialogueState(ttl_seconds=600)使用示例：
state.append("u123", Turn("user", "我的订单在哪？", time.time())) history = state.get_history("u123") # 喂给模型做 context核心实现二：LangChain 版 RAG，让模型读自家文档
- 文档切片：按“标题+两级段落”切，块长 300 token，重叠 50 token。
- 向量库：PGVector + pg14，向量维度 768（bge-base-zh-v1.5）。
- 检索：混合检索（dense 0.7 + bm25 0.3），top-k=6，rerank 用 bge-reranker-large。
代码骨架（省略了异常处理打印）：
from langchain.vectorstores.pgvector import PGVector from langchain.schema import Document from langchain.text_splitter import MarkdownHeaderTextSplitter from langchain.embeddings import HuggingFaceBgeEmbeddings embedding = HuggingFaceBgeEmbeddings( model_name="BAAI/bge-base-zh-v1.5", encode_kwargs={"normalize_embeddings": True} ) CONNECTION_STRING = "postgresql+psycopg2://user:pwd@pg:5432/vectordb" db = PGVector( connection_string=CONNECTION_STRING, embedding_function=embedding, collection_name="faq", ) headers_to_split_on = [ ("#", "Header 1"), ("##", "Header 2"), ] splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on) raw_md = open("wiki.md", encoding="utf-8").read() docs = splitter.split_text(raw_md) db.add_documents(docs)对话时把“用户问题”先向量化，再取回 top-k 段落，拼成 prompt：
def build_prompt(question: str, history: list[dict]) -> str: docs = db.similarity_search(question, k=6) context = "\n".join(d.page_content for d in docs) history_str = "\n".join(f"{t['role']}: {t['content']}" for t in history) return f"""你是客服机器人，仅使用以下资料回答，禁止编造成分。 资料： {context} 历史对话： {history_str} 用户：{question} 客服："""性能优化：让 GPU 不跑满，也能扛 500 并发
批处理对比测试
在 A100-40G 上固定 max_new_tokens=256，换 batch_size：batch 平均吞吐（token/s） 首 token 延迟 1 1,050 180 ms 4 3,800 220 ms 8 6,100 290 ms 16 OOM — 结论：batch=8 是甜点，再大显存炸；线上用 dynamic batching，累积 30 ms 窗口拼批。
LoRA 微调救急
显存不足时，全量微调不现实。peft 库两行代码即可：from peft import LoraConfig, get_peft_model lora_config = LoraConfig( r=16, lora_alpha=32, target_modules=["q_proj", "v_proj"], lora_dropout=0.05, bias="none", task_type="CAUSAL_LM" ) model = get_peft_model(model, lora_config)显存从 37 GB 降到 21 GB，训练速度提升 1.8 倍，下游 BLEU 只掉 0.4，完全可接受。
避坑指南：敏感信息与日志脱敏
正则过滤
手机号、银行卡、身份证，用统一管道先脱敏再进模型：import re def mask_sensitive(text: str) -> str: phone = re.sub(r'1[3-9]\d{9}(?=\D|$)', '<PHONE>', text) id = re.sub(r'\b\d{16,19}\b', '<BANK_CARD>', phone) id_card = re.sub(r'\b\d{6}(19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dX]\b', '<ID>', id) return id_card日志落盘
- 原始日志写本地 tmpfs，定时（5 min）异步批量加密后上传对象存储，减少磁盘 IO。
- 写前再做一次脱敏，key 用 AES-256-GCM，密钥放 KMS，即使运维也看不到明文。
- 保留 30 天自动转冷存，降低 70% 存储费。
延伸思考：增量学习与反馈闭环
上线三个月，我们发现“静态 RAG”开始滞后，新品类 FAQ 每周增 5%。于是搭了半自动闭环：
- 用户点“踩”> 语句自动入“候选池”。
- 运营同学每天花 10 分钟标注“正确答案”，晚上触发增量训练：
- 新数据 1 k 条 + 旧数据随机采样 9 k 条，防止灾难性遗忘。
- 采用 LoRA 合并权重，30 min 完成，灰度 5% 流量，A/B 无回退即全量。
- 模型侧记录“回答-反馈”映射，用 REINFORCE 做轻量 RL，平均奖励 +0.18，仍在小流量实验。
写在最后
这次落地让我深刻体会到：大模型不是“万能胶水”，而是一套需要工程骨架的“乐高”。把状态机、向量检索、批处理、脱敏、增量训练这些模块拼好，才能让大模型在客服场景里真正“说人话、办人事”。如果你也在选型路上，希望这篇笔记能帮你少走几个弯路，少熬几个大夜。
