最近在做一个智能客服系统的升级项目，客户的核心痛点非常明确：客服知识散落在各处，尤其是大量产品文档、FAQ都沉淀在飞书文档里，更新频繁但客服系统无法实时同步，导致机器人经常回答“我不知道”。另一个头疼的问题是，传统的规则或简单模型在意图识别上太僵化，用户稍微换个问法就识别不了。
经过一番折腾，我们最终敲定了一个结合大语言模型（LLM）和飞书文档集成的方案。今天就来聊聊这个实战过程，特别是如何利用AI辅助开发，快速搭建一个“活”的智能客服系统。
1. 为什么是LLM + 飞书文档？
在技术选型上，我们主要对比了两种主流路径：对LLM进行全量微调（Fine-Tuning）和检索增强生成（RAG）架构。
- 全量微调：效果可能很精准，但成本高、周期长。每次知识更新都需要重新训练和部署模型，对于产品文档天天变的业务来说，根本不现实。而且，它很难处理“知识截止日期”之后的新信息。
- RAG架构：这正是我们需要的。它的核心思想是“外挂知识库”。当用户提问时，系统先从外部知识库（比如飞书文档）里找到最相关的片段，再把问题和这些片段一起交给LLM，让它组织语言回答。这样，知识库的更新独立于模型，我们只需要维护好文档源就行。
飞书作为企业内部协作的事实标准，文档丰富且API完善。选择“LLM + 飞书文档”的RAG方案，相当于直接把企业最鲜活的知识资产变成了客服机器人的大脑，实现了动态更新。初步验证，这种架构能将因知识陈旧导致的误判率降低30%以上。
2. 核心实现三步走
整个系统可以拆解为三个核心模块：文档同步、意图理解与对话管理、以及异步处理架构。
2.1 飞书文档同步模块
这是知识的“输入口”。目标是将飞书云文档中的内容，自动、增量地同步到我们的向量数据库中。
首先需要搞定鉴权。飞书开放平台推荐使用OAuth 2.0，但对于企业内部应用或自建应用，使用“自建应用”模式获取tenant_access_token更直接。这里要注意权限范围（bitable:read和wiki:read等）。
import requests import json from typing import Optional class FeishuDocSync: """飞书文档同步器，负责获取Token并拉取文档内容。""" def __init__(self, app_id: str, app_secret: str): self.app_id = app_id self.app_secret = app_secret self.base_url = "https://open.feishu.cn/open-apis" self._token: Optional[str] = None def _get_tenant_access_token(self) -> str: """获取租户访问令牌。内部缓存，过期前重复使用。""" if self._token: # 简单起见，这里假设token未过期。生产环境应校验过期时间。 return self._token url = f"{self.base_url}/auth/v3/tenant_access_token/internal" headers = {"Content-Type": "application/json; charset=utf-8"} payload = {"app_id": self.app_id, "app_secret": self.app_secret} try: response = requests.post(url, headers=headers, data=json.dumps(payload), timeout=10) response.raise_for_status() result = response.json() if result.get("code") == 0: self._token = result["tenant_access_token"] return self._token else: raise Exception(f"Failed to get token: {result.get('msg')}") except requests.exceptions.Timeout: raise Exception("Request to Feishu API timed out.") except requests.exceptions.RequestException as e: raise Exception(f"Network error occurred: {e}") except json.JSONDecodeError as e: raise Exception(f"Failed to parse JSON response: {e}") def fetch_doc_content(self, doc_token: str) -> str: """根据文档token获取纯文本内容。""" token = self._get_tenant_access_token() url = f"{self.base_url}/docx/v1/documents/{doc_token}/raw_content" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json; charset=utf-8" } try: response = requests.get(url, headers=headers, timeout=15) response.raise_for_status() data = response.json() if data.get("code") == 0: # 实际返回结构复杂，这里示意性返回内容 return data.get("data", {}).get("content", "") else: raise Exception(f"API Error: {data.get('msg')}") except requests.exceptions.Timeout: # 生产环境应加入重试机制 raise Exception("Fetch document content timed out.") except (requests.exceptions.RequestException, json.JSONDecodeError) as e: raise Exception(f"Error fetching doc: {e}")拿到文档内容后，需要进行清洗、分块（Chunking），然后通过嵌入模型（Embedding Model）转换成向量，存入像Chroma、Milvus这样的向量数据库。这里的关键是分块策略，太小会失去上下文，太大会影响检索精度，我们根据文档类型（如API文档、FAQ）采用了不同的分块大小。
2.2 Prompt设计模板与对话状态维护
LLM本身没有记忆，多轮对话的状态需要我们自己维护。我们的Prompt模板设计了几部分：
- 系统角色设定：明确告诉AI它是谁。
- 上下文历史：拼接最近几轮的对话记录。
- 检索到的知识：从向量库查找到的相关文档片段。
- 当前用户问题。
- 输出格式指令：要求以特定JSON格式回复，包含“回答”和“下一步建议问题”。
def build_customer_service_prompt(history: list, retrieved_knowledge: list, current_query: str) -> str: """ 构建客服场景的Prompt。 Args: history: 历史对话列表，格式 [("user", "内容"), ("assistant", "内容")] retrieved_knowledge: 检索到的知识片段列表 current_query: 当前用户问题 Returns: 构造好的完整Prompt字符串 """ system_role = "你是一个专业、友善的客服助手。请严格根据提供的知识库信息回答问题。如果知识库中没有相关信息，请如实告知，并引导用户联系人工客服。" # 构建历史对话上下文 history_context = "" for speaker, text in history[-5:]: # 只保留最近5轮 role = "用户" if speaker == "user" else "助手" history_context += f"{role}: {text}\n" # 构建知识上下文 knowledge_context = "\n".join([f"- {k}" for k in retrieved_knowledge]) prompt_template = f""" {system_role} 以下是相关的知识库信息： {knowledge_context} 以下是最近的对话历史： {history_context} 当前用户问题：{current_query} 请根据知识库信息回答用户问题。如果知识库信息不足，请直接说明。 你的回答应该清晰、简洁、有帮助。 请用以下JSON格式输出： {{ "answer": "你的回答内容", "suggested_questions": ["建议问题1", "建议问题2"] }} """ return prompt_template.strip()对话状态（历史记录）我们使用Redis或内存缓存（如cachetools）来维护，以session_id为键。这样就能实现连贯的多轮对话。
2.3 异步处理架构
为了应对并发请求和耗时的文档处理任务，我们采用了异步架构。核心流程如下：
graph TD A[用户提问] --> B(API网关/Web层) B --> C{会话是否存在?} C -->|否| D[创建新会话] C -->|是| E[读取历史记录] D --> F E --> F subgraph F [异步处理核心] F1[向量检索<br/>相关文档] --> F2[构建Prompt] F2 --> F3[调用LLM API] end F3 --> G[解析LLM响应] G --> H[更新对话历史] H --> I[返回回答与建议] I --> J[用户端展示] K[飞书文档变更] --> L(事件订阅/定时任务) L --> M[文档同步服务] M --> N[文本清洗与分块] N --> O[生成向量] O --> P[更新向量数据库]Web层（如FastAPI）接收请求后，快速处理会话状态，然后将耗时的检索和LLM调用丢到异步任务队列（如Celery或asyncio）中，避免阻塞。文档同步则通过订阅飞书事件或定时任务触发，同样走异步流程，确保主服务响应速度。
3. 实战避坑指南
踩过几个坑，这里分享下：
- 飞书API速率限制：飞书API有明确的QPS限制。我们的策略是：1）在同步文档时，加入指数退避的重试机制；2）对于非实时性要求极高的文档，采用分批、间隔同步；3）缓存文档元数据，减少不必要的详情查询。
- LLM输出稳定性：
temperature参数是关键。对于客服这种需要稳定、准确回答的场景，我们设置为0.1-0.3，降低随机性。同时，在Prompt里加入“严格根据知识库”的强指令，并在后端对输出做格式校验和内容安全过滤。 - 敏感信息过滤：即使知识库内部，也可能包含联系方式、内部编号等。我们在文本入库前做了一层过滤：
import re def filter_sensitive_info(text: str) -> str: """简易敏感信息过滤（示例）""" # 过滤手机号（简单示例） text = re.sub(r'1[3-9]\d{9}', '[手机号已屏蔽]', text) # 过滤特定内部项目编号格式，如 PRJ-2024-XXXX text = re.sub(r'PRJ-\d{4}-\w+', '[项目编号已屏蔽]', text) # 可以添加更多规则... return text
4. 性能压测数据
我们模拟了100个并发用户持续提问的场景，并人为为飞书API调用添加了50-200ms的随机延迟（模拟网络波动或API限流）。
下图展示了在引入异步任务队列和缓存后，系统响应时间（P95）的变化。可以看到，虽然单个LLM调用可能较慢，但通过异步化，用户感知的接口响应时间被控制在了可接受的范围内，大部分请求在1.5秒内返回。
（示意图：横轴为并发用户数，纵轴为响应时间(ms)，展示P50、P95线）
5. 代码规范与质量
在团队协作中，代码规范至关重要。我们要求所有Python代码遵循PEP 8，并使用black和isort自动化格式化。关键函数和类必须有清晰的docstring，说明其作用、参数和返回值。异常处理必须覆盖网络请求超时、JSON解析失败、API返回错误码等场景，并为用户提供友好的错误信息，同时记录详细日志供排查。
6. 延伸思考
这个方案的核心思路是“LLM + 实时知识源”。飞书文档只是其中一个优质知识源。这个架构可以很自然地扩展到其他平台：
- 钉钉/企业微信：它们的开放平台也提供了类似的文档、群消息等API。只需替换掉飞书同步模块的API调用部分，适配各自的鉴权和数据格式即可。
- 跨平台同步挑战：真正的挑战在于，当知识分散在多个平台（如飞书、Confluence、GitHub Wiki）时，如何实现统一同步、去重和冲突解决？这需要设计一个更强大的“知识源管理中间层”，定义统一的内容模型，并处理不同平台的更新频率和通知机制差异。
写在最后
通过这次项目，我深刻体会到，AI辅助开发不是用LLM替代开发者，而是让我们能更聚焦于架构设计和业务逻辑整合。像飞书文档同步、Prompt工程、异步流程设计这些工作，仍然需要扎实的工程能力。但LLM的引入，确实让我们能用更少的规则代码，构建出更智能、更灵活的系统。希望这套“LLM+飞书文档”的实战思路，能给你带来一些启发。下一步，我打算试试把GitHub Issues也作为知识源接进来，让客服能回答关于代码库的问题，应该会更有趣。
