最近在帮公司搭建智能客服系统，发现市面上的方案要么太“重”，要么效果不达标。特别是面对用户五花八门的问题时，传统规则引擎经常“卡壳”，而直接调用大模型API又担心成本和响应速度。今天就来聊聊，如何从零开始，用AI技术构建一个既智能又实用的客服系统，类似豆包那样的体验。
一、为什么传统方案越来越吃力？
以前做客服机器人，主流是规则引擎（比如Rasa）或者云服务（比如Dialogflow）。它们在小范围、流程固定的场景下还行，但一旦问题变复杂，短板就暴露了。
- 意图识别僵化：规则引擎严重依赖预先定义的意图和大量的模板例句。用户稍微换种说法，比如把“怎么退款”说成“钱能退回来吗”，可能就识别不出来了。维护这些规则是个无底洞。
- 多轮对话是噩梦：传统的对话状态管理（DST）需要手动设计大量的状态跳转逻辑。一个简单的“查订单-改地址”流程，可能要写几十个状态判断，非常容易出错，且难以扩展。
- 上下文理解能力弱：用户经常在对话中引用上文，比如“上面说的那个订单”、“他（指客服）刚才说的办法”。传统方案很难捕捉这种指代关系，导致对话断裂，用户体验很糟。
正是这些痛点，让基于大语言模型（LLM）的方案成为新的选择。LLM拥有强大的语义理解和生成能力，能更好地处理开放域问题和上下文关联。
二、技术方案怎么选？三大流派横评
决定用AI之后，具体走哪条路？这里对比一下主流的三类方案：
自研规则引擎（以Rasa为代表）
- 优点：数据隐私性好，完全自控，初期对简单场景开发快。
- 缺点：意图识别和对话逻辑需要大量人工标注和设计，维护成本指数级增长，难以处理复杂、开放的对话。
- 适用场景：流程极度标准化、封闭、且对数据安全有极高要求的内部系统。
云服务商方案（如Dialogflow, 腾讯云智聆）
- 优点：开箱即用，提供图形化配置界面，集成方便，通常自带一些预训练模型。
- 缺点：有厂商锁定风险，定制能力受平台限制，按调用量计费，长期使用成本可能较高，数据需上传至厂商服务器。
- 适用场景：追求快速上线、无自研团队、且对数据出境不敏感的中小型业务。
大语言模型API方案（如豆包、文心、GPT等）
- 优点：对话理解能力强，能处理开放域、多轮、指代等复杂场景，开发接口简单，效果上限高。
- 缺点：API调用有延迟和成本，需要设计提示词（Prompt）和上下文管理逻辑来引导模型，存在输出不可控的风险。
- 适用场景：追求高质量对话体验、问题域开放或复杂、有一定技术能力进行工程优化的场景。
简单对比表：
| 维度 | 自研规则引擎 | 云服务方案 | LLM API方案 |
|---|---|---|---|
| 开发成本 | 高（后期维护） | 低（初期） | 中（需工程化） |
| 效果上限 | 低 | 中 | 高 |
| 定制灵活性 | 高 | 低 | 中高 |
| 数据隐私 | 高 | 低 | 中（依赖厂商） |
| 长期成本 | 人力成本高 | 按量付费，可能高 | 按Token付费，需优化 |
对于大多数想打造“类豆包”体验的团队，LLM API + 自建对话管理后台是一个平衡效果、成本和开发周期的选择。下面我们就以这个方向展开。
三、核心实现：用Python搭建对话引擎
我们选择Python作为开发语言，用Flask提供API，核心是构建一个可靠的对话状态管理器。
1. 对话状态管理机（State Machine）实现
这是大脑，负责记住对话历史、理解当前状态、决定下一步动作。我们不依赖LLM做状态管理，而是用更可控的代码逻辑。
# dialogue_state_manager.py import json import time from typing import Dict, List, Optional, Any from dataclasses import dataclass, asdict from abc import ABC, abstractmethod import hashlib @dataclass class DialogueTurn: """表示对话中的一轮交换""" user_input: str system_response: str timestamp: float # 可以扩展意图、实体等信息 intent: Optional[str] = None entities: Optional[Dict] = None def to_dict(self): return asdict(self) class DialogueState: """对话状态核心类""" def __init__(self, session_id: str): self.session_id = session_id self.turns: List[DialogueTurn] = [] # 对话历史 self.context: Dict[str, Any] = { # 自定义上下文信息 "last_intent": None, "slots": {}, # 例如：{"product_name": "手机", "order_id": "12345"} "step": "greeting", # 当前对话步骤 } self.created_at = time.time() self.last_activity = time.time() def add_turn(self, user_input: str, system_response: str, intent: str = None): """添加一轮对话记录，并更新状态""" turn = DialogueTurn(user_input, system_response, time.time(), intent) self.turns.append(turn) self.last_activity = time.time() # 根据意图更新上下文（简化示例） if intent == "query_order": self.context["step"] = "awaiting_order_id" elif intent == "provide_order_id": self.context["step"] = "processing_query" # ... 其他状态转移逻辑 def get_recent_history(self, max_turns: int = 5) -> str: """获取最近N轮对话历史，用于构造LLM的Prompt""" recent = self.turns[-max_turns:] if len(self.turns) > max_turns else self.turns history_text = "" for turn in recent: history_text += f"用户：{turn.user_input}\n系统：{turn.system_response}\n" return history_text.strip() def is_expired(self, timeout_seconds: int = 1800) -> bool: """判断会话是否超时（30分钟无活动）""" return (time.time() - self.last_activity) > timeout_seconds # 状态管理器单例 class DialogueStateManager: _instance = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance.sessions: Dict[str, DialogueState] = {} return cls._instance def get_or_create_session(self, session_id: str) -> DialogueState: """获取或创建对话状态。复杂度：O(1)""" if session_id not in self.sessions or self.sessions[session_id].is_expired(): self.sessions[session_id] = DialogueState(session_id) return self.sessions[session_id] def cleanup_expired_sessions(self): """清理过期会话，防止内存泄漏。复杂度：O(n)""" expired_keys = [k for k, v in self.sessions.items() if v.is_expired()] for k in expired_keys: del self.sessions[k]2. 集成LLM与构建Flask API
接下来，我们集成大模型API（这里以假设的DoubaoClient为例），并暴露服务接口。
# app.py from flask import Flask, request, jsonify import logging from dialogue_state_manager import DialogueStateManager from llm_client import DoubaoClient # 假设的豆包API客户端 from sensitive_filter import AsyncSensitiveFilter # 敏感词过滤 app = Flask(__name__) dsm = DialogueStateManager() llm_client = DoubaoClient(api_key="your_api_key") sensitive_filter = AsyncSensitiveFilter() def build_prompt(user_input: str, history: str, context: dict) -> str: """构造给LLM的提示词。这是效果的关键！""" system_role = """你是一个专业的智能客服助手。请根据对话历史和当前用户问题，提供准确、有帮助、友好的回答。 如果用户询问订单、物流、退款等业务问题，请从已知信息中回答。如果不知道，请引导用户提供更多信息或转接人工。 已知上下文信息： - 当前对话步骤：{step} - 已填写的槽位：{slots} """.format(step=context.get('step'), slots=context.get('slots')) prompt = f"{system_role}\n\n对话历史：\n{history}\n\n当前用户问题：{user_input}\n\n请回复：" return prompt @app.route('/api/chat', methods=['POST']) def chat(): """核心对话接口。处理幂等性，异步过滤敏感词。""" data = request.json session_id = data.get('session_id') user_message = data.get('message') request_id = data.get('request_id') # 用于幂等性校验 if not session_id or not user_message: return jsonify({'error': 'Missing parameters'}), 400 # 1. 幂等性检查（简易版，基于session_id + request_id） # 生产环境应使用Redis等存储已处理的请求ID request_key = f"{session_id}:{request_id}" # ... 此处省略检查逻辑，若已处理则直接返回上次响应 # 2. 异步敏感词检测（非阻塞） sensitive_filter.submit_check(session_id, user_message) # 3. 获取对话状态并更新 state = dsm.get_or_create_session(session_id) # 这里可以加入一个简单的NLU模块来提取意图/实体，或直接让LLM做 # intent, entities = simple_nlu(user_message) # 4. 构建Prompt并调用LLM history_text = state.get_recent_history(max_turns=5) prompt = build_prompt(user_message, history_text, state.context) try: llm_response = llm_client.generate(prompt, max_tokens=500) except Exception as e: logging.error(f"LLM API call failed: {e}") llm_response = "网络开小差了，请稍后再试。" # 5. 更新对话状态并保存 state.add_turn(user_message, llm_response) # 这里可以传入intent # 6. 返回响应 return jsonify({ 'session_id': session_id, 'response': llm_response, 'request_id': request_id }) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=False)四、生产环境必须考虑的坑
代码跑起来只是第一步，要上线稳定服务，下面这些点一个都不能少。
对话日志的幂等性处理
- 问题：网络超时可能导致客户端重试，同一用户请求被处理两次，造成重复回答或状态错乱。
- 方案：要求客户端每个请求带一个唯一
request_id。服务端用Redis等高速缓存记录session_id:request_id和处理结果。收到请求时先查缓存，若已处理则直接返回缓存结果。缓存可设置较短过期时间（如5分钟）。
敏感词过滤的异步检测架构
- 问题：同步过滤会增加响应延迟，影响用户体验。
- 方案：采用生产者-消费者模式。API接口收到消息后，立即将其放入一个消息队列（如Redis Stream/RabbitMQ），然后继续处理逻辑返回响应。后台有独立的消费者进程从队列取出消息，进行详细的敏感词、广告、违规内容检测，并将结果记录到数据库或风控系统。实现业务与风控解耦。
五、从零到一：避坑实战指南
模型冷启动时的语料标注技巧
- 不要只堆数量：初期收集1000条高质量、多样化的用户真实query，远比5000条重复、相似的语料有用。
- 重视负样本：除了标注正确的意图，还要标注一些容易混淆的、边界不清的query，帮助模型（或你的NLU模块）学会区分。
- 利用LLM做数据增强：手头语料不足时，可以用LLM（如ChatGPT）对现有例句进行同义改写、扩充，快速生成更多训练数据。但一定要人工审核，防止引入噪音。
高并发场景下的会话隔离方案
- 内存瓶颈：上述示例的状态管理器存储在进程内存中，多机部署时会话不共享，且重启数据丢失。
- 解决方案：使用Redis等外部存储会话状态。将
DialogueState对象序列化（如JSON或MessagePack）后存入Redis，Key为session_id，并设置合理的TTL（如1小时）。这样任何一台后端实例都能读取和更新同一会话状态，实现了无状态服务的会话有状态化。
六、更进一步：让客服更“懂行”
基本的问答做通了，如何让客服更专业？可以尝试结合知识图谱。
- 思路：将产品手册、常见问题（FAQ）、商品信息等结构化，构建一个小的领域知识图谱。当LLM生成回答时，先通过一个检索模块，从知识图谱中查找最相关的实体和关系，然后将这些结构化信息作为“参考文档”插入到Prompt中。
- 好处：能极大提升回答的准确性和专业性，减少LLM“胡编乱造”的情况。例如，用户问“手机A的电池容量和充电速度”，系统能精准地从知识图谱中检索出“手机A -> 电池容量：5000mAh”、“手机A -> 快充协议：120W”等信息喂给LLM，生成可靠回答。
写在最后
搭建一个智能客服系统，技术选型只是起点，真正的挑战在于工程细节的打磨和效果的持续优化。从简单的状态机到集成LLM，再到处理幂等、过滤敏感词、管理会话状态，每一步都需要结合业务场景仔细设计。
我个人感觉，目前“LLM API + 强工程控制”是一条比较务实的路径。它既利用了LLM强大的语言能力，又通过我们自己的代码逻辑保证了对话的稳定性和可控性。当然，这条路也需要不断调整Prompt、优化上下文管理策略，是一个持续迭代的过程。
希望这篇笔记里分享的思路和代码片段，能给你带来一些启发。下一步，我打算深入研究一下用向量数据库来做长上下文的知识检索，看看能不能让客服的回答更精准。如果你有更好的想法，欢迎一起交流。
