最近在帮公司升级智能客服系统，踩了不少坑，也积累了一些实战经验。传统客服系统在面对复杂多轮对话、意图识别模糊的场景时，常常力不从心，导致客户体验下降，甚至造成业务损失。今天就来聊聊我们基于DeepSeek大模型重构智能客服系统的完整方案，包括架构设计、核心代码实现，以及那些容易踩的“坑”。

1. 传统客服系统的痛点与业务损失

在引入大模型之前，我们调研了公司旧系统的表现，发现几个核心问题：

意图识别准确率低：基于规则和传统NLP模型（如BERT fine-tuning）的意图识别，在开放域问题上准确率只有65%-75%。这意味着每4个客户问题中，就有1个可能被错误分类，需要转人工，直接导致人工客服成本增加约30%。

上下文保持能力弱：传统的对话管理多基于有限状态机（FSM）或填槽（Slot Filling）模式。一旦用户跳出预设流程，比如在询问“手机价格”后突然问“和A型号比哪个拍照好？”，系统往往无法关联上下文，需要用户重复信息。我们的数据显示，因此导致的对话平均轮次增加了2.3轮，用户满意度下降22%。

多语言与复杂问法支持差：对于口语化、带错别字或混合中英文的问句（如“这个pro啥时候deliver？”），传统模型处理效果很差，拒识率高，迫使大量海外用户转向邮件支持，响应时效从秒级降至小时级。

这些缺陷直接反映在业务指标上：我们的分析发现，因智能客服解决率低而流失的潜在客户，约占月度咨询量的5%，折算成营收损失相当可观。这促使我们转向基于大语言模型（LLM）的新方案。

2. 技术选型：为什么是DeepSeek？

我们对比了三种主流方案：基于正则和规则的引擎、基于微调的传统NLP模型（如BERT、RoBERTa）、以及基于大语言模型（LLM）的通用方案（如DeepSeek、GPT系列）。

我们内部搭建了测试集（包含1万条真实客服对话），从三个维度进行了量化对比：

维度	规则/正则引擎	微调BERT模型	DeepSeek-V2 API
意图识别准确率	71.2%	78.5%	94.7%
平均响应延迟	< 50ms	120-200ms	350-800ms (网络依赖)
上下文关联能力	无/很弱	中等	强
开发与维护成本	高（规则膨胀）	中（需标注数据、训练）	低（Prompt驱动）
泛化能力	差	中等	优秀

关键结论：

1. DeepSeek在意图识别准确率上优势明显，尤其在处理“未见过”的问法时，凭借其强大的语言理解能力，准确率远超传统模型。
2. 响应延迟是LLM的短板，但通过异步处理、流式响应和缓存可以极大优化用户体验，感知延迟并不高。
3. 成本考量：虽然DeepSeek API调用有费用，但省去了昂贵的标注数据、模型训练和GPU服务器运维成本，总体TCO（总拥有成本）在业务量增长后反而更低。

因此，我们选择了DeepSeek作为核心的对话理解与生成引擎，同时保留部分高频、确定性高的规则作为快速通道（Fast Path），兼顾准确性与性能。

3. 系统架构设计：四层弹性架构

我们的目标是构建一个能支撑10K+ TPS（每秒事务数）、高可用的智能客服系统。整体架构分为四层：

接入层（Gateway Layer）：

• 使用 Nginx 做负载均衡和SSL终结。
• API Gateway（采用Kong）负责路由、限流（Rate Limiting）、认证鉴权。所有请求必须携带有效的JWT Token。
• 将用户请求发布到异步消息队列，实现接入层与业务层的解耦，应对流量洪峰。

对话引擎层（Dialogue Engine Layer）：

• 这是系统的“大脑”，核心是对话状态机（Dialogue State Machine, DSM），负责管理整个会话的生命周期。
• 意图识别模块：首先调用DeepSeek API进行零样本（Zero-Shot）或小样本（Few-Shot）意图分类。对于明确意图（如“查询订单”），直接触发后续流程；对于模糊或复杂意图，进入多轮对话管理。
• 对话管理模块：维护对话状态（Dialogue State），包括用户意图、已填写的槽位（Slots）、对话历史等。它决定系统下一步是“询问澄清”、“调用知识库”还是“执行动作”。
• 响应生成模块：根据对话状态，或从知识库检索答案，或格式化业务数据，或调用DeepSeek生成自然语言回复。

知识库与数据层（Knowledge & Data Layer）：

• 向量知识库：使用Milvus或Chroma存储产品文档、FAQ的向量嵌入（Embedding）。当用户问题需要事实性答案时，对话引擎会先从此检索最相关的片段，再连同片段一起发给DeepSeek生成最终回复，这是减少“模型幻觉”的关键。
• 业务数据库：存储用户信息、订单数据等，供对话引擎查询。
• 缓存（Redis）：缓存会话状态、高频问答对、API调用结果，大幅降低延迟和成本。

监控与运维层（Monitoring & Ops Layer）：

• 使用 Prometheus 收集各项指标（QPS、延迟、错误率、DeepSeek Token消耗）。
• Grafana 用于可视化仪表盘。
• ELK Stack（Elasticsearch, Logstash, Kibana）集中收集和分析日志，便于调试和审计。

异步消息队列选型（Kafka vs Pulsar）： 我们最终选择了 Apache Pulsar，主要基于以下几点考量：

• 架构更现代：Pulsar采用计算（Broker）与存储（Bookie）分离的架构，扩容更灵活。Kafka的Broker同时负责计算和存储，扩容时需要数据重平衡，影响性能。
• 多租户和命名空间：Pulsar原生支持多租户，非常适合我们SaaS化的规划。Kafka需要自行在Topic命名上实现。
• Geo-Replication：Pulsar的跨地域复制配置更简单，对我们未来部署多区域机房很有吸引力。
• 延迟与吞吐：两者都能满足我们的万级TPS需求。Pulsar在保证低延迟的同时，提供了更稳定的尾部延迟（Tail Latency）。

当然，Kafka的生态更成熟，社区更大。如果你的团队对Kafka非常熟悉，且场景相对简单，Kafka依然是优秀的选择。

4. 核心实现代码详解

4.1 对话状态机（Dialogue State Machine）实现

对话状态机是管理多轮对话的核心。我们使用Python 3.10+实现，并利用Redis进行状态持久化。

import json
import time
import uuid
from enum import Enum
from typing import Dict, Any, Optional, List
from dataclasses import dataclass, asdict, field
from redis import Redis

class DialogueState(Enum):
    """对话状态枚举"""
    GREETING = "greeting"  # 欢迎
    IDENTIFYING_INTENT = "identifying_intent"  # 识别意图
    COLLECTING_SLOTS = "collecting_slots"  # 收集信息槽位
    EXECUTING_ACTION = "executing_action"  # 执行动作（如查询）
    PROVIDING_ANSWER = "providing_answer"  # 提供答案
    CLARIFYING = "clarifying"  # 澄清中
    ENDED = "ended"  # 对话结束

@dataclass
class Slot:
    """信息槽位定义"""
    name: str
    value: Any = None
    required: bool = True
    prompt: str = ""  # 询问用户时的提示语

@dataclass
class DialogueContext:
    """对话上下文，代表一次完整的会话状态"""
    session_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    current_state: DialogueState = DialogueState.GREETING
    intent: Optional[str] = None
    slots: Dict[str, Slot] = field(default_factory=dict)
    history: List[Dict[str, str]] = field(default_factory=list)  # 对话历史记录
    created_at: float = field(default_factory=time.time)
    last_updated: float = field(default_factory=time.time)
    # 其他业务相关字段...

class DialogueStateManager:
    """对话状态管理器"""
    
    def __init__(self, redis_client: Redis, ttl_seconds: int = 1800):
        """
        初始化
        :param redis_client: Redis客户端
        :param ttl_seconds: 会话状态在Redis中的存活时间（秒），默认30分钟
        """
        self.redis = redis_client
        self.ttl = ttl_seconds
        
    def get_context(self, session_id: str) -> Optional[DialogueContext]:
        """从Redis获取对话上下文"""
        data = self.redis.get(f"dialogue:ctx:{session_id}")
        if not data:
            return None
        ctx_dict = json.loads(data)
        # 重建上下文对象（简化处理，实际需更复杂的反序列化）
        ctx = DialogueContext()
        ctx.session_id = ctx_dict.get('session_id')
        ctx.current_state = DialogueState(ctx_dict.get('current_state'))
        ctx.intent = ctx_dict.get('intent')
        # ... 反序列化其他字段
        return ctx
    
    def save_context(self, ctx: DialogueContext) -> None:
        """保存对话上下文到Redis"""
        ctx.last_updated = time.time()
        ctx_dict = asdict(ctx)
        # 枚举类型需要转换为字符串存储
        ctx_dict['current_state'] = ctx.current_state.value
        key = f"dialogue:ctx:{ctx.session_id}"
        self.redis.setex(key, self.ttl, json.dumps(ctx_dict))
        
    def transition_state(self, 
                         session_id: str, 
                         new_state: DialogueState,
                         **kwargs) -> DialogueContext:
        """
        执行状态转移，并更新上下文
        :param session_id: 会话ID
        :param new_state: 目标状态
        :param kwargs: 需要更新的上下文字段
        :return: 更新后的对话上下文
        """
        ctx = self.get_context(session_id)
        if not ctx:
            # 新会话，创建上下文
            ctx = DialogueContext(session_id=session_id)
        
        ctx.current_state = new_state
        ctx.last_updated = time.time()
        
        # 更新其他传入的字段
        for key, value in kwargs.items():
            if hasattr(ctx, key):
                setattr(ctx, key, value)
                
        # 记录状态转移历史（可选，用于调试）
        ctx.history.append({
            'timestamp': time.time(),
            'from': ctx.current_state.value,  # 注意：这里简化了，实际应记录前一个状态
            'to': new_state.value,
            'intent': ctx.intent
        })
        
        self.save_context(ctx)
        return ctx
    
    def handle_timeout(self, session_id: str) -> None:
        """处理会话超时，清理资源或发送超时提示"""
        ctx = self.get_context(session_id)
        if ctx and (time.time() - ctx.last_updated > self.ttl):
            # 强制转移到结束状态
            self.transition_state(session_id, DialogueState.ENDED)
            # 可以在这里触发一个超时通知或清理动作
            print(f"Session {session_id} timed out.")

这个状态机管理器提供了基础的会话状态持久化和超时处理能力。在实际业务中，transition_state方法会与具体的业务逻辑（如下面的DeepSeek调用）结合，根据用户输入和当前状态决定下一个状态。

4.2 集成DeepSeek API与Prompt Engineering

与DeepSeek API的交互是系统的核心。我们封装了一个服务类，并精心设计了Prompt以减少模型幻觉。

import httpx
from typing import List, Dict
import asyncio

class DeepSeekClient:
    """DeepSeek API客户端封装"""
    
    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com"):
        self.api_key = api_key
        self.base_url = base_url
        self.client = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=30.0
        )
    
    async def chat_completion(self, 
                             messages: List[Dict[str, str]],
                             model: str = "deepseek-chat",
                             temperature: float = 0.1,  # 低温度，减少随机性
                             max_tokens: int = 500) -> Dict[str, Any]:
        """
        调用DeepSeek聊天补全API
        """
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": False  # 非流式，如需流式响应可改为True
        }
        
        try:
            resp = await self.client.post("/chat/completions", json=payload)
            resp.raise_for_status()
            return resp.json()
        except httpx.HTTPStatusError as e:
            # 处理HTTP错误
            print(f"HTTP error occurred: {e}")
            raise
        except Exception as e:
            # 处理其他错误
            print(f"Error calling DeepSeek API: {e}")
            raise
    
    async def classify_intent(self, user_query: str, possible_intents: List[str]) -> str:
        """
        使用DeepSeek进行意图分类（Few-Shot示例）
        """
        # 构建Few-Shot Prompt
        few_shot_examples = """
        请将用户的查询分类到以下意图之一：{intent_list}。
        
        示例：
        用户：我的订单123456到哪里了？
        意图：查询物流状态
        
        用户：我想退货，怎么操作？
        意图：申请退货
        
        用户：这款手机有现货吗？
        意图：查询库存
        
        现在请对以下查询进行分类：
        用户：{query}
        意图：
        """.format(intent_list=", ".join(possible_intents), query=user_query)
        
        messages = [
            {"role": "system", "content": "你是一个专业的客服意图分类助手。请只输出意图名称，不要输出其他任何解释。"},
            {"role": "user", "content": few_shot_examples}
        ]
        
        response = await self.chat_completion(messages, temperature=0)
        intent = response['choices'][0]['message']['content'].strip()
        
        # 确保返回的意图在候选列表中，否则返回'其他'
        return intent if intent in possible_intents else '其他'
    
    async def generate_answer_with_knowledge(self, 
                                             user_query: str, 
                                             knowledge_snippets: List[str],
                                             conversation_history: List[str]) -> str:
        """
        结合知识库片段生成回答，有效减少幻觉
        """
        # 构建强约束性的Prompt
        system_prompt = """你是一个专业的客服助手。请严格根据提供的“参考信息”来回答用户问题。
        参考信息是唯一可信的来源。如果参考信息不足以回答用户问题，请明确告知用户“根据现有资料，我无法回答这个问题”，并建议其联系人工客服。
        绝对不要编造参考信息中没有的内容。
        
        参考信息：
        {knowledge}
        
        当前对话历史（最近3轮）：
        {history}
        """.format(
            knowledge="\n---\n".join(knowledge_snippets),
            history="\n".join(conversation_history[-3:]) if conversation_history else "无"
        )
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_query}
        ]
        
        response = await self.chat_completion(messages, temperature=0.1)
        return response['choices'][0]['message']['content'].strip()
    
    async def close(self):
        """关闭HTTP客户端"""
        await self.client.aclose()

# 使用示例
async def main():
    client = DeepSeekClient(api_key="your_api_key_here")
    
    # 示例1：意图识别
    intents = ["查询订单", "咨询产品", "投诉建议", "售后服务", "其他"]
    user_question = "我昨天买的手机屏幕碎了，能保修吗？"
    detected_intent = await client.classify_intent(user_question, intents)
    print(f"识别到的意图: {detected_intent}")
    
    # 示例2：结合知识库生成回答
    relevant_knowledge = [
        "产品保修政策：非人为损坏享受一年保修，人为损坏（如屏幕碎裂）不在保修范围内。",
        "屏幕维修服务：可前往官方授权服务中心付费维修，或购买碎屏险的用户可免费更换一次。"
    ]
    history = ["用户：你好", "客服：您好，有什么可以帮您？"]
    answer = await client.generate_answer_with_knowledge(user_question, relevant_knowledge, history)
    print(f"生成的回答: {answer}")
    
    await client.close()

if __name__ == "__main__":
    asyncio.run(main())

Prompt Engineering技巧总结：

1. 系统指令（System Prompt）要明确：清晰定义助手的角色、职责和限制。
2. 使用Few-Shot示例：对于分类等任务，提供少量示例能显著提升效果。
3. 提供参考依据：将知识库检索结果作为“参考信息”放入Prompt，并严格要求模型基于此回答，这是对抗幻觉最有效的方法之一。
4. 控制温度（Temperature）：对于客服这种需要确定性的场景，将温度设为较低值（如0.1），减少回答的随机性。
5. 结构化输出要求：明确要求模型以特定格式（如只输出意图名称）输出，便于程序解析。

5. 生产环境考量

5.1 压力测试方案与结果

系统上线前，我们使用JMeter进行了全面的压力测试。目标是验证在1万并发用户下的系统表现。

测试场景：

• 模拟用户从发起会话、多轮对话（平均5轮）到结束的完整流程。
• 混合请求类型：70%简单QA（命中缓存或知识库），30%复杂会话（需要调用DeepSeek API）。

JMeter配置关键点：

• 使用 HTTP Request Sampler 模拟API调用。
• 使用 JSON Extractor 和 Regular Expression Extractor 从响应中提取session_id等参数，用于后续请求，模拟有状态会话。
• 使用 Synchronizing Timer 实现瞬间高并发。
• 通过 Backend Listener 将结果发送到InfluxDB，再通过Grafana展示。

压测结果（在8核16G的API服务器节点上）：

• 平均响应时间（Average Response Time）：在1万并发下，整体平均响应时间为320ms。其中，直接返回缓存结果的请求<50ms，调用DeepSeek API的请求在800-1200ms（主要耗时在网络和模型推理）。
• 吞吐量（Throughput）：系统稳定在约9500 TPS。
• 错误率（Error Rate）：在持续30分钟的压测中，错误率低于0.1%，主要为超时错误（设置了2秒超时）。
• 资源利用率：CPU平均使用率75%，内存使用平稳。DeepSeek API侧，根据其提供的监控，我们的调用平稳，未触发限流。

优化措施：

• 实现请求合并与批处理：对于短时间内相似的用户问题，合并后一次性调用DeepSeek API，减少调用次数。
• 引入多级缓存：使用Redis缓存高频问答、会话状态，甚至对某些DeepSeek的确定性回答进行短期缓存（设置较短TTL）。
• 配置弹性伸缩（Auto Scaling）：根据CPU利用率和请求队列长度，自动增减处理节点。

5.2 安全防护

1. 认证鉴权（JWT）： 所有客户端请求必须在Header中携带有效的JWT Token。我们在API Gateway（Kong）层面统一验证。

# 示例：简单的JWT验证逻辑（实际使用应依赖成熟库如PyJWT）
import jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key-here"  # 应从环境变量读取

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm="HS256")
    return encoded_jwt

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.ExpiredSignatureError:
        return None  # Token过期
    except jwt.InvalidTokenError:
        return None  # Token无效

2. 对话内容过滤： 为了防止用户输入恶意内容或触发模型生成不当回复，我们在调用DeepSeek API前进行内容过滤。

import re

class ContentFilter:
    def __init__(self):
        # 1. 敏感词过滤（可使用AC自动机，见下文）
        self.sensitive_patterns = [
            r"(?i)badword1|badword2",  # 示例
            r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b",  # 简单电话号码过滤示例
        ]
        # 2. 提示注入攻击检测模式
        self.prompt_injection_patterns = [
            r"(?i)ignore (previous|above) instructions",
            r"(?i)now you are going to play a role of",
            r"system.*prompt.*user",
        ]
        
    def filter_text(self, text: str) -> tuple[bool, str]:
        """
        过滤文本
        :return: (是否通过, 过滤后的文本或拒绝原因)
        """
        # 检查敏感词
        for pattern in self.sensitive_patterns:
            if re.search(pattern, text):
                return False, "内容包含敏感信息"
        
        # 检查提示注入
        for pattern in self.prompt_injection_patterns:
            if re.search(pattern, text):
                return False, "检测到潜在指令注入攻击"
        
        # 其他检查，如长度限制、特殊字符比例等
        if len(text) > 1000:
            return False, "输入内容过长"
            
        return True, text.strip()

6. 实战避坑指南

6.1 解决大模型“幻觉”（Hallucination）

幻觉是指模型生成看似合理但事实上不正确或无关的信息。在客服场景，这是致命的。

我们的解决方案：

1. 知识库优先（Retrieval-Augmented Generation, RAG）：如4.2节代码所示，强制模型基于检索到的知识片段回答。
2. 设计抗幻觉Prompt模板：

   ANTI_HALLUCINATION_PROMPT_TEMPLATE = """
   你是一个准确、可靠的客服助手。请严格遵守以下规则：
   
   规则1：你的回答必须严格基于“参考信息”部分提供的事实。参考信息是唯一可信来源。
   规则2：如果参考信息中没有足够信息来完全回答用户问题，你必须说：“根据现有资料，我无法完全确认这个问题。建议您 [提供通用建议，如‘联系我们的客服专员’或‘查看官网最新公告’]。”
   规则3：绝对不要猜测、假设或编造任何参考信息中没有的细节，包括但不限于数字、日期、政策条款、人物、地点。
   规则4：如果用户的问题与参考信息完全无关，请礼貌地表示无法回答，并引导用户提出与业务相关的问题。
   
   参考信息：
   {knowledge_context}
   
   对话历史（最近几句）：
   {conversation_history}
   
   用户问题：{user_question}
   
   请根据以上规则和参考信息回答：
   """
   

3. 后处理校验：对于模型生成的、涉及关键事实（如价格、日期、政策条款）的回复，可以尝试用另一个小的、确定性的分类模型或规则进行事实性校验。

6.2 会话粘性与Redis键设计

在分布式部署中，确保同一用户会话的请求被路由到同一个服务实例（或能访问到相同的会话状态）很重要。

Redis键设计策略：

# 核心思想：使用用户唯一标识（如user_id）或会话ID（session_id）作为键的一部分
class RedisKeyManager:
    PREFIX = "cs:"  # cs = customer service
    
    @classmethod
    def session_context(cls, session_id: str) -> str:
        return f"{cls.PREFIX}ctx:{session_id}"
    
    @classmethod
    def user_sessions(cls, user_id: str) -> str:
        # 存储某用户的所有活跃会话ID，用于清理或查询
        return f"{cls.PREFIX}user:{user_id}:sessions"
    
    @classmethod
    def rate_limit(cls, user_id: str, endpoint: str) -> str:
        # 限流键
        return f"{cls.PREFIX}rate:{user_id}:{endpoint}"
    
    @classmethod
    def cached_response(cls, query_hash: str) -> str:
        # 缓存常见问题的回答，query_hash可以是用户问题的MD5
        return f"{cls.PREFIX}cache:resp:{query_hash}"

会话粘性保持：

• 方案A（客户端保持）：客户端在首次请求后，将服务端返回的session_id存储在本地（如Cookie或LocalStorage），后续请求均携带此ID。服务端通过session_id从Redis读取统一上下文。
• 方案B（负载均衡器保持）：配置负载均衡器（如Nginx）的ip_hash或sticky session，将同一IP的请求转发到固定后端。但这对移动端或动态IP用户不友好。
• 推荐方案：采用方案A。服务端设计为无状态的，会话状态完全存储在共享缓存（Redis）中。这样既实现了粘性，又保证了后端服务的水平扩展能力。

6.3 高效敏感词过滤：AC自动机实现

正则表达式在词量很大时效率较低。我们使用AC自动机（Aho–Corasick算法）实现高效的敏感词过滤。

from ahocorasick import Automaton  # 需要 pip install pyahocorasick

class SensitiveWordFilter:
    def __init__(self, sensitive_words_file: str):
        self.automaton = Automaton()
        self.load_words(sensitive_words_file)
        
    def load_words(self, filepath: str):
        """从文件加载敏感词，构建AC自动机"""
        with open(filepath, 'r', encoding='utf-8') as f:
            for idx, word in enumerate(f):
                word = word.strip()
                if word:
                    # 添加词，并可以存储一个值（这里存词本身）
                    self.automaton.add_word(word, (idx, word))
        self.automaton.make_automaton()
        
    def contains_sensitive(self, text: str) -> bool:
        """检查文本是否包含敏感词"""
        for end_pos, (idx, original_word) in self.automaton.iter(text):
            # 只要找到一个就返回True
            return True
        return False
    
    def replace_sensitive(self, text: str, replace_char: str = "*") -> str:
        """替换文本中的敏感词为指定字符"""
        sensitive_positions = []
        for end_pos, (idx, original_word) in self.automaton.iter(text):
            start_pos = end_pos - len(original_word) + 1
            sensitive_positions.append((start_pos, end_pos))
        
        # 将位置合并，避免重叠替换导致问题
        sensitive_positions.sort()
        merged_positions = []
        for start, end in sensitive_positions:
            if not merged_positions or start > merged_positions[-1][1]:
                merged_positions.append([start, end])
            else:
                merged_positions[-1][1] = max(merged_positions[-1][1], end)
        
        # 从后往前替换，避免索引变化
        result_chars = list(text)
        for start, end in reversed(merged_positions):
            result_chars[start:end+1] = replace_char * (end - start + 1)
            
        return ''.join(result_chars)

# 使用示例
filter = SensitiveWordFilter("sensitive_words.txt")
text = "这是一个包含不良词语的测试句子。"
if filter.contains_sensitive(text):
    cleaned_text = filter.replace_sensitive(text)
    print(f"过滤后: {cleaned_text}")
else:
    print("文本安全。")

AC自动机能在O(n)的时间复杂度内完成多模式匹配，非常适合敏感词过滤场景。我们将它部署在API Gateway或对话引擎的入口处，作为第一道防线。

写在最后

从传统规则引擎切换到基于DeepSeek的智能客服系统，对我们来说是一次显著的升级。准确率的提升直接带来了客户满意度和客服效率的改善。当然，这个过程也充满挑战，尤其是在确保响应速度、控制成本和维护回答准确性之间的平衡。

最大的一个开放性问题，也是我们持续在探索的：如何平衡大模型成本与响应速度？

DeepSeek API的调用费用和响应延迟（尤其是复杂推理）是两大成本。我们的策略是：

1. 分层处理：简单、高频问题走缓存或规则引擎；复杂、开放性问题才调用DeepSeek。
2. 缓存一切可缓存的：对模型生成的、确定性高的回答进行适当时间的缓存。
3. 考虑混合模型：对于某些垂直领域，是否可以用一个更小、更快的本地模型（如微调的7B参数模型）处理大部分问题，只在必要时求助DeepSeek？
4. 监控与优化：持续监控Token消耗和响应时间，优化Prompt以减少不必要的Token使用，并设置预算告警。

这条路还在继续，希望我们踩过的坑和总结的方案能给你带来一些启发。如果你有更好的思路或实践经验，欢迎一起交流。