最近在做一个智能客服项目客户那边对并发量和响应速度要求特别高。传统的基于规则或者简单意图匹配的客服系统在高并发下响应延迟大而且多轮对话的上下文理解能力基本为零用户体验很差。经过一番调研和折腾我们最终基于Ollama这个大模型本地化部署工具搭建了一套还算不错的方案吞吐量相比之前提升了3倍多。今天就来分享一下整个架构设计和性能优化的过程希望能给有类似需求的同学一些参考。1. 为什么不用传统方案痛点分析最开始我们用的是规则引擎关键词匹配的客服系统上线没多久就暴露了几个硬伤意图识别准确率低用户的问题千奇百怪规则很难覆盖所有情况。比如用户问“我的订单怎么还没到”规则可能只匹配“订单”和“到”但无法理解这是“物流查询”意图还是“催单”意图更别说结合上下文了。多轮对话是灾难传统方案几乎没有上下文保持能力。用户问“帮我查一下订单”客服回复“请提供订单号”用户再回复“123456”这时系统很可能已经忘了上一轮对话是关于“查询订单”的导致对话断裂。高并发下性能瓶颈当大量用户同时涌入时即使是简单的规则匹配和数据库查询也会因为I/O阻塞导致响应时间飙升用户体验急剧下降。维护成本高每增加一个新的业务场景或问题类型都需要人工去添加和维护一大堆规则费时费力而且容易出错。这些痛点让我们下定决心必须引入具备真正语义理解能力的大模型。2. 技术选型Ollama vs. ChatGPT API决定用大模型后摆在面前的主要有两个选择使用云服务商如OpenAI的ChatGPT API或本地部署开源模型。我们重点对比了Ollama和ChatGPT API。成本考量ChatGPT API按Token收费对于高并发的客服场景长期来看是一笔不小的持续支出。而且数据需要出境存在合规风险。Ollama一次性投入主要在硬件GPU服务器上。模型开源免费部署在本地内网无持续调用费用数据完全自主可控。从长期运营成本看Ollama优势明显。响应延迟与稳定性ChatGPT API依赖网络存在网络波动和API服务稳定性的风险。虽然通常很快但在网络不佳或对方服务抖动时延迟不可控。Ollama模型部署在内网服务器网络延迟极低毫秒级。响应速度主要取决于本地GPU的算力稳定性完全由自己掌控。微调与定制灵活性ChatGPT API提供微调接口但过程相对黑盒且对微调数据的格式和规模有要求定制深度有限。Ollama支持多种开源模型如Llama 2, Mistral, CodeLlama等可以自由选择适合客服场景的模型。最关键的是可以基于业务数据历史客服对话记录进行全量微调或高效的LoRA (Low-Rank Adaptation)微调让模型更懂我们的业务术语和流程这是提升准确率的关键。综合来看对于需要高并发、低延迟、数据安全且希望深度定制模型的企业级客服场景Ollama本地化部署是更优解。我们最终选择了Mistral 7B模型它在效果和推理速度上取得了很好的平衡。3. 核心架构设计为了支撑高并发我们的架构不能只是一个简单的模型调用包装必须考虑异步、缓存和任务队列。FastAPI 异步网关使用FastAPI构建HTTP接口层。它的异步特性非常适合I/O密集型操作如等待模型推理结果。它负责接收用户请求、基础验证、对话状态管理并将耗时的模型推理任务下发。Redis 对话状态缓存每个会话Session的对话历史是核心上下文。我们将session_id作为Key将压缩后的对话历史列表作为Value存储在Redis中。利用Redis的高性能和过期机制可以快速读写上下文并自动清理过期会话。Celery 异步任务队列模型推理是计算密集型任务耗时较长几百毫秒到几秒。如果直接在HTTP请求线程中同步调用会迅速阻塞所有工作线程。我们使用Celery作为分布式任务队列将推理任务异步化。FastAPI接口收到请求后立即向Celery发送一个推理任务并返回一个任务ID。前端可以通过轮询或WebSocket来获取任务结果。这样API网关的响应速度极快不会因为模型推理慢而受影响。4. 关键代码实现下面展示一些核心代码片段关键逻辑都加了注释。4.1 带JWT认证的FastAPI路由与对话处理from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials import jwt from pydantic import BaseModel from typing import List, Optional import uuid import redis.asyncio as redis from celery_app import inference_task # 导入Celery任务 app FastAPI(title智能客服API) security HTTPBearer() SECRET_KEY your-secret-key-here # 生产环境应从配置读取 ALGORITHM HS256 # 连接Redis池用于缓存对话历史 redis_pool redis.ConnectionPool.from_url(redis://localhost:6379/0, decode_responsesTrue) redis_client redis.Redis(connection_poolredis_pool) class Message(BaseModel): role: str # user or assistant content: str class ChatRequest(BaseModel): session_id: Optional[str] None # 首次请求为空后续携带 message: str stream: bool False # 是否流式输出 def verify_token(credentials: HTTPAuthorizationCredentials Depends(security)): JWT令牌验证依赖项 token credentials.credentials try: payload jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) return payload # 通常包含用户ID等信息 except jwt.PyJWTError: raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail无效或过期的令牌, ) app.post(/v1/chat) async def chat_completion( request: ChatRequest, user_info: dict Depends(verify_token) # 依赖注入进行认证 ): 处理用户聊天请求 # 1. 生成或验证session_id session_id request.session_id or str(uuid.uuid4()) # 2. 从Redis获取历史对话 history_key fchat_history:{session_id} raw_history await redis_client.lrange(history_key, 0, -1) # 获取列表 # 简单反序列化实际存储可能用json history: List[Message] [Message.parse_raw(msg) for msg in raw_history] # 3. 将用户新消息加入历史并进行压缩减少token user_msg Message(roleuser, contentrequest.message) history.append(user_msg) compressed_history compress_conversation(history) # 压缩算法见下文 # 4. 将压缩后的历史存回Redis设置过期时间如30分钟 serialized_history [msg.json() for msg in compressed_history[-10:]] # 只保留最近10轮 await redis_client.delete(history_key) # 先删除旧的 if serialized_history: await redis_client.rpush(history_key, *serialized_history) await redis_client.expire(history_key, 1800) # 30分钟过期 # 5. 异步调用Celery任务进行模型推理 task inference_task.delay( session_idsession_id, conversation [{role: m.role, content: m.content} for m in compressed_history], user_iduser_info.get(sub) ) # 6. 立即返回告知客户端通过task_id查询结果 return { session_id: session_id, task_id: task.id, status: processing } app.get(/v1/task/{task_id}) async def get_task_result(task_id: str): 查询异步任务结果 from celery_app import app as celery_app task_result celery_app.AsyncResult(task_id) if task_result.state SUCCESS: # 任务成功返回模型生成的回复 result task_result.get() # 这里可以再将助手回复存入Redis历史 return {status: success, response: result} elif task_result.state FAILURE: return {status: failure, error: str(task_result.info)} else: return {status: task_result.state} # PENDING, STARTED等4.2 对话历史压缩算法直接传递全部历史对话会给模型带来巨大的token开销增加成本和延迟。我们需要压缩历史。def compress_conversation(history: List[Message], max_tokens: int 2000) - List[Message]: 简单的对话历史压缩算法。 策略优先保留最近几轮对话如果还是超长则对最早的几轮进行摘要。 from some_tokenizer import tokenizer # 假设有tokenizer compressed [] total_tokens 0 # 从最新对话开始处理逆序 for msg in reversed(history): msg_tokens len(tokenizer.encode(msg.content)) if total_tokens msg_tokens max_tokens: compressed.insert(0, msg) # 在头部插入保持顺序 total_tokens msg_tokens else: # 如果超出限制对当前这条实际上是最早的之一进行摘要 if msg.role user: # 简单摘要取开头一部分。生产环境可用小模型或规则摘要。 summarized_content msg.content[:150] ...[已摘要] else: summarized_content msg.content[:100] ...[已摘要] summarized_msg Message(rolemsg.role, contentsummarized_content) sum_tokens len(tokenizer.encode(summarized_content)) # 检查摘要后是否能加入 if total_tokens sum_tokens max_tokens: compressed.insert(0, summarized_msg) total_tokens sum_tokens else: # 即使摘要后也放不下了直接中断只保留能放下的部分 break return compressed4.3 超时熔断机制实现为了防止某个模型调用异常如Ollama服务卡死拖垮整个系统需要实现熔断。import time from functools import wraps from typing import Callable import asyncio class CircuitBreaker: 一个简单的熔断器实现 def __init__(self, failure_threshold: int 5, recovery_timeout: int 60): self.failure_threshold failure_threshold # 连续失败次数阈值 self.recovery_timeout recovery_timeout # 恢复时间秒 self.failures 0 self.state CLOSED # CLOSED, OPEN, HALF-OPEN self.last_failure_time None def call(self, func: Callable, *args, **kwargs): if self.state OPEN: # 检查是否过了恢复时间 if time.time() - self.last_failure_time self.recovery_timeout: self.state HALF-OPEN print(熔断器进入半开状态尝试放行一个请求) else: raise Exception(Circuit breaker is OPEN. Service unavailable.) try: result func(*args, **kwargs) # 调用成功重置失败计数 if self.state HALF-OPEN: self.state CLOSED self.failures 0 print(熔断器关闭服务恢复) elif self.state CLOSED: self.failures 0 return result except Exception as e: self.failures 1 self.last_failure_time time.time() print(f调用失败失败次数: {self.failures}) if self.failures self.failure_threshold: self.state OPEN print(f熔断器打开服务暂停) raise e # 使用示例包装Ollama模型调用函数 ollama_breaker CircuitBreaker() def call_ollama_with_breaker(prompt: str): with ollama_breaker: # 这里调用实际的Ollama API设置超时 response requests.post(http://localhost:11434/api/generate, json{model: mistral, prompt: prompt}, timeout10.0) # 设置10秒超时 response.raise_for_status() return response.json()5. 性能优化实战架构搭好了代码写完了性能到底怎么样必须压测使用Locust进行压力测试 Locust是一个用Python写的开源负载测试工具可以用代码定义用户行为非常灵活。# locustfile.py from locust import HttpUser, task, between import jwt import time class ChatUser(HttpUser): wait_time between(1, 3) # 用户思考时间1-3秒 host http://localhost:8000 def on_start(self): # 模拟登录获取Token self.token self.generate_token() self.headers {Authorization: fBearer {self.token}} self.session_id None def generate_token(self): payload {sub: test_user, exp: time.time() 3600} return jwt.encode(payload, your-secret-key-here, algorithmHS256) task def send_message(self): # 模拟发送消息并查询结果 chat_data { session_id: self.session_id, message: 请问我的订单发货了吗, stream: False } # 1. 发送聊天请求 with self.client.post(/v1/chat, jsonchat_data, headersself.headers, catch_responseTrue) as resp: if resp.status_code 200: data resp.json() self.session_id data[session_id] task_id data[task_id] # 2. 轮询获取结果简单模拟实际可用长轮询/WebSocket max_retries 10 for i in range(max_retries): time.sleep(0.5) # 等待0.5秒 task_resp self.client.get(f/v1/task/{task_id}, headersself.headers) if task_resp.status_code 200: task_data task_resp.json() if task_data[status] success: # print(f收到回复: {task_data[response]}) break else: resp.failure(f请求失败: {resp.status_code})运行locust -f locustfile.py然后在浏览器打开http://localhost:8089设置模拟用户数和每秒产生用户数就可以看到实时的RPS每秒请求数、响应时间、失败率等关键指标。Batch Size对吞吐量的影响 Ollama的API本身可能不支持批量推理。但我们的Celery worker可以并行处理多个任务。我们测试了不同并发worker数量下系统的整体吞吐量QPS。发现并不是worker越多越好因为受限于GPU的算力。在我们的场景Mistral 7B RTX 4090下启动4个worker进程每个进程内使用异步能达到吞吐量和延迟的最佳平衡。同时要监控GPU利用率确保没有成为瓶颈。GPU显存监控方案 使用nvidia-smi命令可以实时查看显存占用。我们写了一个简单的监控脚本定期采集数据并发送到监控系统如Prometheus。# 使用nvidia-smi查询显存使用情况 nvidia-smi --query-gpumemory.used,memory.total --formatcsv,noheader,nounits也可以在Python中使用pynvml库来编程获取。import pynvml pynvml.nvmlInit() handle pynvml.nvmlDeviceGetHandleByIndex(0) # GPU 0 mem_info pynvml.nvmlDeviceGetMemoryInfo(handle) print(f显存使用: {mem_info.used / 1024**2:.2f} MB / {mem_info.total / 1024**2:.2f} MB)监控显存有助于我们发现内存泄漏例如模型热更新未清理旧模型或异常的长上下文积累。6. 避坑指南与经验总结在实际部署和运行中我们踩过不少坑这里分享几个重要的对话日志脱敏存储 客服对话可能包含手机号、地址、订单号等敏感信息。在将对话日志存入数据库或ELK用于分析前必须进行脱敏。我们采用异步处理策略Celery任务完成后将原始对话和模型回复发送到一个专门的“日志处理队列”。另一个消费者服务从队列中取出消息使用正则表达式和NLP实体识别工具如一些开源NER模型识别敏感信息并用***替换然后再存储。这样既不影响主流程响应速度又保证了数据安全。模型热更新的内存泄漏 为了在不重启服务的情况下更新模型例如上线一个LoRA微调后的新版本我们实现了热更新。但最初发现更新几次后GPU显存会不断增长最终OOM。原因是旧的模型权重没有被正确地从GPU显存和Python内存中释放。解决方案在加载新模型之前必须显式地清理旧模型。对于PyTorch需要调用model.cpu()然后del model最后可能还需要torch.cuda.empty_cache()。更稳妥的方式是将模型服务单独进程化例如用gRPC通过重启子进程来更新模型。敏感词过滤的异步处理 直接在主推理流程中进行复杂的敏感词过滤如调用外部API或查大表会增加延迟。我们的做法是“先回复后过滤”。模型生成回复后立即返回给用户。同时将生成的回复内容发送到另一个低优先级的“审核队列”。审核服务异步检查内容如果发现严重违规可以在后台进行记录、告警甚至通过其他渠道如短信、客服人工介入联系用户。这是一种服务降级和保证核心体验的策略。结语与开放性问题通过这套基于Ollama的架构我们成功将智能客服系统的吞吐量提升了300%以上平均响应时间控制在1.5秒内并且具备了优秀的上下文理解能力。本地化部署也让数据安全和长期成本得到了保障。当然没有完美的方案我们仍在持续优化。最后抛出一个开放性问题供大家思考在资源有限的情况下如何平衡模型精度使用更大、更复杂的模型与响应延迟/吞吐量之间的关系是选择量化Quantization来缩小模型加快推理还是采用模型蒸馏Distillation获得更小的学生模型或者是在架构上做文章比如为不同复杂度的查询路由到不同规模的模型这可能是智能客服乃至所有大模型应用都会面临的经典权衡。