最近在做一个智能客服系统的升级项目，之前用的方案响应慢、维护起来也头疼。趁着这个机会，我研究并落地了基于 DeepSeek API 的方案，效果挺不错，响应速度上去了，成本也降下来了。今天就把整个实践过程梳理一下，分享给有类似需求的同学。

传统客服系统，尤其是基于规则或者早期 NLP 模型的，痛点很明显。一是响应慢，用户一个问题过来，可能要经过好几层规则匹配或者复杂的模型推理，高峰期排队严重，用户体验差。二是意图识别不准，稍微换个说法可能就匹配不上，或者答非所问，导致大量问题需要转人工。三是扩展性差，每增加一个业务场景，就得写一堆新规则或者重新训练模型，运维成本非常高。

在技术选型上，我对比了几个主流的大模型 API。OpenAI 的 GPT 系列能力很强，但在中文场景下的语料和语义理解深度，有时候感觉不如专门优化过的中文模型，而且对于国内团队来说，网络延迟和合规性也是需要考虑的因素。国内的一些大厂 API，比如文心一言，中文理解确实不错，但在我们内部测试中，DeepSeek 在长文本理解、多轮对话连贯性以及性价比方面表现更均衡。特别是它的 API 文档清晰，调用方式简单，对于快速构建原型和上线非常友好。

下面我重点讲讲核心的实现部分。首先是一个完整的、健壮的 API 调用封装类。这个类不仅要完成基础的请求，还要处理好鉴权、错误重试、超时控制等。

import aiohttp
import asyncio
import json
import time
from typing import Optional, Dict, Any, List
from tenacity import retry, stop_after_attempt, wait_exponential
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DeepSeekChatClient:
    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com/v1/chat/completions"):
        self.api_key = api_key
        self.base_url = base_url
        self.session: Optional[aiohttp.ClientSession] = None
        self._request_count = 0
        self._total_latency = 0.0

    async def __aenter__(self):
        self.session = aiohttp.ClientSession(
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            timeout=aiohttp.ClientTimeout(total=30)
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    async def chat_completion_async(
        self,
        messages: List[Dict[str, str]],
        model: str = "deepseek-chat",
        temperature: float = 0.7,
        max_tokens: Optional[int] = 1024,
        stream: bool = False
    ) -> Dict[str, Any]:
        """异步调用聊天补全API，内置重试机制"""
        if not self.session:
            raise RuntimeError("Client session not initialized. Use async context manager.")

        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
        }
        if max_tokens:
            payload["max_tokens"] = max_tokens
        if stream:
            payload["stream"] = stream

        start_time = time.time()
        try:
            async with self.session.post(self.base_url, json=payload) as response:
                response.raise_for_status()
                result = await response.json()
                latency = time.time() - start_time

                # 性能监控埋点
                self._request_count += 1
                self._total_latency += latency
                logger.info(f"API Request completed. Latency: {latency:.3f}s, Avg: {self._total_latency/self._request_count:.3f}s")

                return result
        except aiohttp.ClientError as e:
            logger.error(f"Network error during API call: {e}")
            raise
        except json.JSONDecodeError as e:
            logger.error(f"Failed to decode JSON response: {e}")
            raise
        finally:
            # 确保性能数据记录
            pass

    def get_performance_metrics(self) -> Dict[str, float]:
        """获取性能指标"""
        avg_latency = self._total_latency / self._request_count if self._request_count > 0 else 0
        return {
            "total_requests": self._request_count,
            "total_latency": self._total_latency,
            "average_latency": avg_latency
        }

智能客服的核心是多轮对话，这就需要状态管理。我们不能把整个对话历史每次都全量发给 API，那样既浪费 Token（成本），也可能很快触及上下文长度上限。我的做法是维护一个固定长度的对话窗口，并提炼关键信息作为“对话摘要”。

1. 对话状态管理类设计：这个类负责维护一个会话（Session）的生命周期。它会存储用户和助手的对话历史，但只保留最近 N 轮，更早的对话会被压缩成一段摘要。
2. 上下文窗口与摘要压缩：我设置了一个阈值，比如最近 10 轮对话保留原样。当对话轮数超过这个阈值，最早的几轮对话会被提取出关键意图和结论，合并成一段简短的文本摘要，放在系统提示词（System Prompt）或用户消息的开头，以此来实现“长记忆”。
3. 系统提示词工程：系统提示词（System Prompt）是引导模型行为的关键。我会在这里明确 AI 的角色（例如“你是某公司的专业客服助手”）、回答风格（热情、简洁、专业）、知识范围（“请仅基于已知信息回答，不知道的请明确告知”）以及一些输出格式要求。一个好的系统提示能显著提升回复的准确性和可控性。

class DialogueStateManager:
    def __init__(self, max_history_turns: int = 10, summary_model_client=None):
        self.max_history_turns = max_history_turns
        self.history: List[Dict[str, str]] = []  # 存储完整的 messages 结构
        self.summary = ""
        self.client = summary_model_client  # 可选：用于生成摘要的轻量模型客户端

    def add_interaction(self, user_input: str, assistant_response: str):
        """添加一轮完整的交互到历史记录"""
        self.history.append({"role": "user", "content": user_input})
        self.history.append({"role": "assistant", "content": assistant_response})

        # 如果历史记录过长，则进行压缩摘要
        if len(self.history) > self.max_history_turns * 2:  # 每轮包含user和assistant两条
            self._compress_history()

    def _compress_history(self):
        """压缩早期的对话历史为摘要"""
        # 简单策略：将最早的两轮对话（共4条消息）内容提取关键信息，合并到summary中
        to_compress = self.history[:4]
        compressed_text = " | ".join([f"{msg['role']}:{msg['content'][:50]}..." for msg in to_compress])
        self.summary = (self.summary + " [Earlier: " + compressed_text + "] ").strip()
        # 移除已压缩的历史
        self.history = self.history[4:]

    def get_messages_for_api(self, system_prompt: str) -> List[Dict[str, str]]:
        """生成准备发送给API的messages列表"""
        messages = [{"role": "system", "content": system_prompt}]
        if self.summary:
            # 将摘要作为一条独立的系统或用户消息插入
            messages.append({"role": "system", "content": f"对话历史摘要：{self.summary}"})
        messages.extend(self.history[-self.max_history_turns*2:])  # 加入最近的完整历史
        return messages

接下来是提升吞吐量的关键：异步处理。在客服场景，尤其是网页在线客服，会有大量并发的会话请求。使用同步请求会阻塞事件循环，严重限制系统能同时服务的用户数。我用 aiohttp 和 asyncio 实现了全异步的请求处理。

4. 异步请求池：利用 asyncio.gather 或 asyncio.Semaphore 来控制并发度，避免对 API 后端造成过大压力。上述封装好的 DeepSeekChatClient 本身就支持异步调用。
5. 性能压测对比：我使用 locust 工具进行了压测。在相同的服务器配置下，同步版本的 QPS（每秒查询率）大概在 15 左右，而异步版本轻松达到了 120+。更重要的是，异步版本的 P99 延迟（99% 的请求响应时间）更加稳定，在并发 100 的情况下，同步版本 P99 飙升到 2 秒以上，而异步版本保持在 800 毫秒以内。
6. 缓存策略设计：对于高频、通用的用户问题（例如“营业时间”、“密码重置流程”），其答案是固定的。我为这些问题的标准回答建立了缓存。当用户问题进来后，先经过一个简单的语义相似度匹配（可以用 TF-IDF 或轻量级句子向量），如果命中缓存，直接返回结果，完全绕过大模型 API，这能节省大量资源和时间。
7. 限流与熔断：为了保护 DeepSeek API 不被突发流量打垮，也为了控制成本，必须在调用层实现限流。我使用了 asyncio.Semaphore 来限制同时进行的 API 请求数。同时，集成一个简单的熔断器（Circuit Breaker），当 API 错误率（如超时、5xx错误）超过一定阈值时，自动熔断一段时间，直接返回降级内容（如“服务繁忙，请稍后再试”），避免雪崩。

在实际开发中，还遇到一些需要特别注意的“坑”。

8. 中文分词与长度计算：大模型的计费和理解能力都与 Token 数量有关。英文 Token 通常是一个词或子词，而中文 Token 可能是一个字或一个词。DeepSeek 的 Tokenizer 对中文的处理方式需要留意。在截断过长上下文时，不能简单按字符数截断，最好使用模型对应的分词器（如 tiktoken 或 transformers 库中的对应方案）来准确计算和截断，否则可能截断在半个词中间，导致语义混乱。
9. 上下文长度限制处理：这是多轮对话的硬约束。除了前面提到的“摘要压缩”法，还可以采用更智能的策略，比如只保留与当前用户问题最相关的历史对话片段（这需要引入向量检索技术），或者当对话实在过长时，友好地提示用户开启一个新会话。
10. 敏感词与合规过滤：作为对外服务的客服，内容安全至关重要。不能完全依赖模型的自律。我们必须在 AI 回复返回给用户之前，加一层后过滤（Post-filter）。可以维护一个敏感词库进行匹配，也可以使用专门的内容安全 API 进行审核。一旦发现违规内容，立即替换为预定义的安全回复。

最后，聊聊我的体会和展望。通过这套基于 DeepSeek API 的异步智能客服架构，我们确实实现了响应速度的大幅提升（压测显示超过 300%）和运维成本的降低。大模型时代的客服系统，正在从“关键词匹配”和“固定流程”向“深度语义理解”和“个性化对话”演进。

未来的方向，我觉得有几个点可以深入：

• 精准知识库检索（RAG）：将公司内部的产品文档、FAQ 做成向量知识库。用户提问时，先从中检索出最相关的几段信息，再连同问题和这些信息一起发给大模型，让模型生成基于精准知识的回答，避免“胡言乱语”。
• 工作流与工具调用：当用户需要执行查订单、退换货等实际操作时，客服 AI 不应该只停留在“告知流程”，而应该能通过安全的方式调用内部工具 API，真正帮用户完成操作，实现从“问答”到“办事”的跨越。
• 多模态支持：如果未来 API 支持，客服可以处理用户上传的图片（如产品故障图、截图），实现更强大的支持能力。

整个落地过程虽然有不少细节要打磨，但 DeepSeek API 的稳定性和易用性让基础搭建变得非常顺畅。希望这篇笔记能为你提供一些可行的思路和代码参考。如果你在实践过程中有新的发现或更好的优化方案，也欢迎一起交流。