背景痛点:为什么传统智能客服总“答非所问”
过去一年,我至少帮三家客户做过“客服机器人”改造,最常见的吐槽有三句:
- “用户换种问法,机器人就懵圈”——意图识别准确率低于 80%,全靠人工兜底。
- “多问两句,机器人就失忆”——多轮对话状态维护靠 Redis 里一串神秘字符串,维护成本极高。
- “高峰期响应 3 s 起步,用户直接挂断”——异步链路太长,前端等到超时。
归根结底,传统 NLU(Rasa、Dialogflow 之类) pipeline 太长:分词 → 意图 → 槽位 → 回复,每一环都可能掉链子。只要有一个环节训练数据不足,整条链路就“塌方”。于是我们把目光投向了“大模型端到端”方案——FastGPT。
技术对比:FastGPT vs Rasa vs Dialogflow
| 维度 | FastGPT | Rasa 3.x | Dialogflow CX |
|---|---|---|---|
| 成本 | 按 token 计费,无最低消费;可本地部署省调用费 | 开源免费,但 GPU/人力运维成本高 | Google 定价 + 区域限制,阶梯价 |
| 定制性 | 系统提示 + 外挂知识库,改 prompt 即可 | 训练 NLU + Core,需标注数据 | 通过 webhook 扩展,但受平台限制 |
| 响应速度 | 首 token 200 ms 级(流式),端到端 1 s 内 | 本地 GPU 100 ms 级,但需维护特征工程 | 平均 800 ms,高峰期被限流 |
| 中文体验 | 原生支持,标点/口语化好 | 需额外语料,否则 OOV 严重 | 中文支持一般,需手动加同义词 |
一句话总结:
“想快速上线、又不想背一屁股 GPU 债”——FastGPT 是目前最平衡的方案。
核心实现:30 行代码搞定 FastGPT 客户端
下面给出可直接拷到项目里的fastgpt_client.py,已在线上稳定跑两周,支持:
- Bearer 鉴权自动续期
- SSE 流式响应,边返回边打印
- 指数退避重试,网络抖动也不炸
# fastgpt_client.py from __future__ import annotations import os import time import httpx from typing import AsyncIterator, Dict, Any FASTGPT_URL = os.getenv("FASTGPT_URL", "https://fastgpt-api.example.com/v1/chat") FASTGPT_KEY = os.getenv("FASTGPT_KEY", "") class FastGPTClient: """线程安全的 FastGPT 异步客户端.""" def __init__(self, timeout: int = 10) -> None: self.timeout = timeout self._client = httpx.AsyncClient(timeout=timeout) async def stream_chat( self, messages: list[Dict[str, str]], max_retry: int = 3, backoff: float = 0.5, ) -> AsyncIterator[str]: """流式对话,逐句 yield 模型回复. Args: messages: 历史消息,格式 OpenAI-like. max_retry: 最大重试次数. backoff: 退避系数. Yields: 逐 token 字符串. """ headers = {"Authorization": f"Bearer {FASTGPT_KEY}"} payload = {"model": "fastgpt-3.5", "messages": messages, "stream": True} for attempt in range(1, max_retry + 1): try: async with self._client.stream( "POST", FASTGPT_URL, json=payload, headers=headers ) as resp: resp.raise_for_status() async for line in resp.aiter_lines(): if line.startswith("data: "): chunk = line.removeprefix("data: ") if chunk == "[DONE]": return yield chunk return except Exception as e: # noqa: BLE001 if attempt == max_retry: raise await asyncio.sleep(backoff * (2 ** (attempt - 1)))调用示例:
import asyncio async def main(): client = FastGPTClient() messages = [{"role": "user", "content": "我的快递到哪了?"}] async for token in client.stream_chat(messages): print(token, end="", flush=True) if __name__ == "__main__": asyncio.run(main())对话状态机:让机器人“记得”上一句
大模型虽然自带上下文,但生产环境必须“可追踪、可回滚”。我习惯用“状态机 + 快照”模式:
- 把每轮对话打包成
DialogueTurn - 整个 Session 以
List[DialogueTurn]形式落库 - 超过 4 轮自动摘要,防止 token 爆炸
from dataclasses import dataclass, asdict from typing import List import json @dataclass class DialogueTurn: role: str content: str timestamp: float class SessionManager: """负责上下文保持与压缩.""" def __init__(self, max_turns: int = 8) -> None: self.max_turns = max_turns self.turns: List[DialogueTurn] = [] def add(self, role: str, content: str) -> None: self.turns.append(DialogueTurn(role, content, time.time())) if len(self.turns) > self.max_turns: # TODO: 可接入 LLM 自动摘要 self.turns = self.turns[-self.max_turns // 2 :] def to_openai(self) -> list[dict[str, str]]: return [{"role": t.role, "content": t.content} for t in self.turns] def to_json(self) -> str: return json.dumps([asdict(t) for t in self.turns], ensure_ascii=False)这样,即便用户 30 分钟后回来追问,也能从 DB 里恢复整段上下文,实现“断点续聊”。
生产考量:QPS 从 30 到 300 的优化秘籍
- 连接池
把httpx.AsyncClient提成单例,减少 TCP 三次握手。 - 批处理
相同问题高频出现时,用本地 LRU 缓存答案,FastGPT 只做“缓存未命中”的兜底。 - 流式输出
前端采用EventSource,首包 200 ms 内返回,用户体感提升一倍。
压测数据(4C8G,单进程):
- 单并发:平均响应 820 ms
- 50 并发:平均响应 1.05 s,QPS ≈ 45
- 加缓存后:平均响应 320 ms,QPS ≈ 290
安全性:别让大模型“说漏嘴”
- 输入过滤
正则 + 敏感词树双保险,把政治、暴力、色情关键词先挡在门外。 - 输出脱敏
手机号、身份证统一打码****,用presid库即可。 - 提示词加固
系统提示里加一句“若用户索要他人隐私,请拒绝。”能挡住 80% 社工试探。
避坑指南:三天踩出来的三个血坑
- 令牌耗尽
现象:HTTP 402 返回,{"error": "quota exceeded"}
解决:提前一周订阅套餐 + 每日定时任务查余额,低于 20% 发钉钉告警。 - 超时配置不当
现象:Nginx 504,但模型其实还在跑
解决:外层网关 timeout 设 60 s,FastGPT 内部 max_tokens 控制,双边对齐。 - 上下文丢失
现象:用户刷新网页后机器人“失忆”
解决:SessionManager 每次add()后同步写 Redis,并设置 24 h TTL;前端带session_idcookie。
代码规范小结
- 统一
black格式化,行宽 88 - 所有公开函数补全类型标注与 docstring
- 单元测试覆盖 > 80%,核心 client 用
pytest-httpxmock,CI 跑 3.9/3.10/3.11 三版本
开源实验仓库
完整代码与 Docker-Compose 模板已放在 GitHub:
https://github.com/yourname/fastgpt-cs-boilerplate
欢迎提 Issue 交流。
结尾思考
当大模型带来“拟人”体验的同时,成本与速度的天平仍在摇摆:
“如果业务场景 QPS 再涨 10 倍,你会优先压缩 prompt 长度,还是投入更高并发实例?”
期待在评论区看到你的实践答案。