基于飞书云文档与LLM的智能客服系统架构设计与工程实践

基于飞书云文档与LLM的智能客服系统架构设计与工程实践

摘要：本文针对传统客服系统响应慢、知识库更新滞后等痛点，提出基于飞书云文档与LLM的智能客服解决方案。通过飞书开放平台实时同步知识库，结合LLM的意图识别与生成能力，实现客服响应速度提升300%。文章详细解析系统架构设计、飞书API集成技巧、对话状态机实现，并提供可复用的Python代码示例与性能压测数据。

1. 传统客服的三大顽疾

1. 知识库更新慢
   旧系统用 Confluence 手工维护，运营同学改完页面后，还要提工单给研发，再走一遍“导出 PDF → 上传 CMS → 重建索引”流程，平均滞后 2～3 天。用户问“新活动规则”，客服只能甩链接，体验瞬间拉胯。

2. 多轮对话无状态
   早期关键词机器人只能“一句一问”，用户说“我订单丢了，昨天付的款”，机器人回“请提供订单号”，用户再回“12345”，机器人却忘了上下文，又得从头问一遍，转化率直接掉 30%。

3. 并发高就卡死
   大促峰值 1200 QPS，旧系统把 FAQ 全放 MySQL，like 查询把 CPU 打满，P99 延迟飙到 8 s，客服页面刷不出答案，只能电话回呼，成本翻倍。

2. 技术选型：为什么不是 Confluence 或 Notion？

结论：飞书在“实时回调 + 高并发 + 免费额度”三点碾压，最适合做知识源。

LLM 选型

• GPT-3.5-turbo：成本 0.002 $/1 k tokens，512 上下文足够 FAQ 场景。
• GPT-4：推理慢 3 倍，贵 15 倍，仅在对准确率要求 >95% 的灰色兜底场景启用。
  最终策略：3.5 做第一轮回答，置信度 <0.8 再走 4 二次校验，成本降 70%。

3. 系统总览

┌--------------┐ 回调推送 ┌--------------┐ │ 飞书云文档 │---deltaEvent--->│ 知识网关 │──┐ └--------------┘ └--------------┘ │ ▼ ┌--------------┐ 检索请求 ┌--------------┘ │ 客服小程序 │<---answerJSON----│ LLM 服务 │ └--------------┘ └--------------┘

关键指标

• 端到端延迟：≤ 800 ms（含网络）
• 知识更新 SLA：≤ 10 s
• 幻觉率：≤ 2%（用 RAG + 自检双保险）

4. 飞书文档实时同步实现

4.1 监听入口

飞书在文档“保存”时会触发document.version.change事件，把doc_token、revision_id、change_type推给我们。网关收到后只回 200，不处理业务，避免超时。

4.2 增量内容拉取

用GET /open-apis/doc/v1/{doc_token}/raw_content拿最新 Markdown，再跟本地快照做 Git-style diff，拿到+/-块。好处：

1. 只把变更段落重新 embedding，节省 80% 向量调用。
2. 删除段落直接按doc_id+block_id硬删，避免脏数据。

4.3 冲突控制

飞书回调可能乱序，网关用 Redis 队列做“revision_id”单调递增校验，发现回退直接丢弃。

5. 对话状态机（Python 版）

状态机解决“多轮对话”痛点，把一次咨询拆成 4 个状态：
INIT → AWAIT_ORDER → AWAIT_REASON → DONE

# state_machine.py import asyncio, time from enum import Enum, auto from dataclasses import dataclass class State(Enum): INIT = auto() AWAIT_ORDER = auto() AWAIT_REASON = auto() DONE = auto() @dataclass class Context: user_id: str state: State = State.INIT order_id: str = "" reason: str = "" expire_at: float = 0 # 时间戳，超时清除 class DialogMachine: """单用户状态机，线程安全（协程）""" def __init__(self, ttl: int = 300): self._ctx: dict[str, Context] = {} self.ttl = ttl # 5 分钟没交互就回收 async def transit(self, uid: str, inp: str) -> str: now = time.time() ctx = self._ctx.get(uid) # 超时或首次 if not ctx or now > ctx.expire_at: ctx = Context(user_id=uid, expire_at=now + self.ttl) self._ctx[uid] = ctx if ctx.state == State.INIT and "订单" in inp: ctx.state = State.AWAIT_ORDER ctx.expire_at = now + self.ttl return "请提供订单号：" if ctx.state == State.AWAIT_ORDER: ctx.order_id = inp.strip() ctx.state = State.AWAIT_REASON ctx.expire_at = now + self.ttl return "请问遇到什么问题？" if ctx.state == State.AWAIT_REASON: ctx.reason = inp ctx.state = State.DONE # 这里调用下游 LLM 生成答案 answer = await self._call_llm(ctx) self._ctx.pop(uid, None) # 清理状态 return answer # 默认兜底 return "我没理解，请再说一次" async def _call_llm(self, ctx: Context) -> str: # 伪代码，后面给出完整提示词 prompt = f"用户订单{ctx.order_id}，问题：{ctx.reason}，请用友好语气回答。" return await llm_chat(prompt)

异常处理

• 任何状态抛出asyncio.TimeoutError自动回到INIT并提示“会话已重置”。
• 限流熔断：下游 LLM 返回 429 时，状态机不迁移，直接返回“服务繁忙，稍后再试”。

6. LLM 提示词模板（RAG 版）

You are「小飞客服」，语气亲切，回答不超过 80 字。 上下文知识： {chunks} 用户问题：{question} 如果上下文无法回答，请说“暂无资料，已转人工”。 不要编造优惠金额、活动日期。

{chunks}由向量库召回 Top3，相似度阈值 ≥0.78，不足就触发“转人工”兜底，减少幻觉。

7. 核心代码：飞书 API 封装 + 异步消息处理

# feishu.py import aiohttp, asyncio, time from typing import List, Dict class FeishuClient: """飞书 OpenAPI 轻量封装，自动换 token、限流重试""" def __init__(self, app_id: str, app_secret: str): self.app_id = app_id self.app_secret = app_secret self._tenant_access_token = "" self._expire = 0 self._session: aiohttp.ClientSession | None = None async def _ensure_token(self): """线程安全换 token，提前 60 s 刷新""" now = time.time() if self._expire < now + 60: url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal" body = {"app_id": self.app_id, "app_secret": self.app_secret} async with self._session.post(url, json=body) as r: r.raise_for_status() js = await r.json() self._tenant_access_token = js["["tenant_access_token"] self._expire = now + js["expire"] - 60 async def get_markdown(self, doc_token: str) -> str: await self._ensure_token() url = f"https://open.feishu.cn/open-apis/doc/v1/{doc_token}/raw_content" headers = {"Authorization": f"Bearer {self._tenant_access_token}"} async with self._session.get(url, headers=headers) as r: if r.status == 429: # 简单退避 await asyncio.sleep(1) return await self.get_markdown(doc_token) r.raise_for_status() data = await r.json() return data["content"] async def __aenter__(self): self._session = aiohttp.ClientSession() return self async def __aexit__(self, exc_type, exc, tb): if self._session: await self._session.close()

异步消息处理（基于 asyncio + Redis Stream）

# worker.py import asyncio, json, aioredis, logging from feishu import FeishuClient from state_machine import DialogMachine logging.basicConfig(level=logging.INFO) log = logging.getLogger("worker") REDIS_DSN = "redis://localhost:6379/0" GROUP = "csbot" CONSUMER = "worker-1" STREAM = "feishu_events" async def main(): feishu = FeishuClient(app_id="xxx", app_secret="yyy") machine = DialogMachine() redis = aioredis.from_url(REDIS_DSN) await redis.xgroup_create(STREAM, GROUP, id="$", mkSTREAM=True) while True: # 拉 1 条，阻塞 1 s msgs = await redis.xreadgroup(STREAM, GROUP, CONSUMER, count=1, block=1000) if not msgs: continue for _, data in msgs: try: event = json.loads(data[b"event"]) doc_token = event["doc_token"] md = await feishu.get_markdown(doc_token) # TODO: 解析 diff 并更新向量库 log.info("synced %s", doc_token) except Exception as e: log.exception("error %s", e) if __name__ == "__main__": asyncio.run(main())

8. 压测数据

用 k6 脚本模拟 1 k 并发 WebSocket 长连接，持续 5 min，结果如下：

800 QPS 是拐点，再往上需横向扩容 LLM 服务，或把向量库从单节点 Milvus 切到分布式集群。

9. 敏感信息过滤

正则 + 双层黑名单，线上跑 30 天无漏报。

import re PHONE_RE = re.compile(r"1[3-9]\d{9}") IDCARD_RE = re.compile(r"\d{15}|\d{18}") BLACK_WORDS = {"微信", "QQ", "支付宝", "转账"} def mask_sensitive(text: str) -> str: text = PHONE_RE.sub("📞", text) text = IDCARD_RE.sub("", text) for w in BLACK_WORDS: text = text.replace(w, "▓▓") return text

10. 避坑指南

1. 飞书 API 限流
   文档接口共享 500 次/秒，大促期间先被秒杀。解决：

   • 把“全量拉取”改“增量 diff”，调用量降 90%。
   • 被 429 后指数退避，最大 8 s，防止死循环。

2. LLM 幻觉检测
   采用“SelfCheckGPT”思路：把答案再让模型判断“能否从上下文推出”，置信度 <0.85 就打回人工。上线后幻觉率从 7% 降到 2%。

3. 向量库内存暴涨
   初期用 OpenAI 1536 维，一千万段要 60 GB。后换国产 BGE-M3 768 维，量化到 fp16，内存砍半，检索掉点 <1%。

11. 小结与展望

整个项目从立项到灰度共 6 周，飞书当知识源省掉 CMS 后台，LLM 负责“说人话”，状态机保证“不尬聊”。线上运行三个月，客服人效提升 40%，夜间 80% 会话无需人工。下一步准备把“图片+表格”也送进多模态向量模型，让机器人也能看懂活动海报，继续给运营同学减负。

如果你也在用飞书办公，不妨把文档直接当知识库，少建一套 CMS，真香。