最近在做一个智能客服项目用上了大语言模型LLM。想法很美好但一上线就遇到了现实问题用户稍微一多系统响应就慢得像蜗牛GPU内存也蹭蹭往上涨成本根本扛不住。经过一番折腾我们摸索出了一套从架构到部署的优化方案效果还不错吞吐量提了3倍GPU内存省了40%。今天就把这套实战经验整理出来希望能帮到有类似困扰的朋友。1. 背景痛点为什么你的LLM客服会“卡顿”刚开始我们采用了一个非常直接的架构一个Flask/FastAPI服务每次用户请求过来就加载模型进行一次完整的推理。这个方案在Demo阶段没问题但一到生产环境问题全暴露了。实时性差每个请求都是串行处理前面的人问个复杂问题后面的人就得干等着。P99延迟99%的请求响应时间轻松突破10秒用户体验极差。并发能力弱GPU虽然算力强但一次只服务一个请求利用率极低。想象一下8个昂贵的GPU核心大部分时间在“围观”一个核心干活。资源浪费严重每个请求都会单独加载一次模型的权重和KV缓存Key-Value Cache用于记录历史对话的键值对避免重复计算。用户一多内存瞬间就被重复的缓存占满导致OOM内存溢出。问题的核心在于我们把LLM当成了普通的Web服务来设计没有考虑到其“计算密集”和“状态保持”的特性。传统Web请求是无状态的、轻量的而LLM推理是重量的、有状态的需要维护上下文。2. 技术选型效率提升的三板斧针对上述痛点我们评估了几个主流优化方向模型量化把模型参数从高精度如FP32转换为低精度如INT8/INT4。这能大幅减少模型体积和内存占用推理速度也能提升。代价是可能会有轻微的性能损失如回答质量下降。对于客服场景回答的稳定性和准确性比极致创意更重要量化是性价比极高的选择。动态批处理把短时间内到来的多个用户请求智能地打包成一个批次送给GPU一次性计算。这能极大提升GPU的利用率。关键在于“动态”因为每个用户的输入长度不同需要高效的填充和调度策略。流式响应不等模型生成完整回答就一边生成一边把文字片段推送给前端。这能极大改善用户感知上的延迟用户看到第一个字的时间首字延迟会大大提前。我们的策略是量化打底批处理提效流式响应改善体验。三者结合能系统性地解决问题。3. 核心实现从零搭建高效推理服务3.1 使用TGI部署量化模型我们选择了 HuggingFace 的Text Generation Inference框架。它原生支持模型量化、动态批处理和流式输出开箱即用。首先拉取并运行TGI服务。这里我们使用量化后的 Llama2-7B-Chat 模型。# 使用Docker部署是最简单的方式 docker run --gpus all -p 8080:80 -v /path/to/models:/data ghcr.io/huggingface/text-generation-inference:latest --model-id /data/Llama-2-7b-chat-hf --quantize bitsandbytes-nf4 --max-batch-total-tokens 102400这条命令做了几件事--quantize bitsandbytes-nf4: 使用NF4格式进行4比特量化内存占用降至约1/4。--max-batch-total-tokens 102400: 设置批次处理的总token上限TGI会根据这个限制动态组合请求。3.2 实现基于Redis的请求队列TGI本身有调度器但对于超大规模并发或者想实现更复杂的优先级队列如VIP用户优先可以引入Redis作为缓冲队列。我们的设计是Web API层接收请求后将任务信息用户ID、问题、时间戳存入Redis Stream。独立的推理Worker从Stream中消费任务调用TGI服务并将结果写回Redis再由API层返回给用户。这样实现了请求的异步化与削峰填谷。# producer.py - API服务端负责接收和排队请求 import redis import json import uuid from typing import Dict, Any from pydantic import BaseModel redis_client redis.Redis(hostlocalhost, port6379, decode_responsesTrue) STREAM_KEY llm:inference:requests class ChatRequest(BaseModel): user_id: str question: str session_id: str None # 用于维护对话会话 def submit_request(request: ChatRequest) - str: 提交一个推理请求到队列返回任务ID task_id str(uuid.uuid4()) message { task_id: task_id, user_id: request.user_id, question: request.question, session_id: request.session_id or request.user_id } # 使用Redis Stream存储支持多消费者和消息持久化 redis_client.xadd(STREAM_KEY, message) return task_id3.3 编写异步推理WorkerWorker是系统的核心它需要高效地从队列取任务调用TGI并处理结果。# worker.py - 异步推理工作进程 import asyncio import aiohttp import redis import json from typing import AsyncGenerator TGI_SERVER_URL http://localhost:8080/generate_stream redis_client redis.Redis(hostlocalhost, port6379, decode_responsesTrue) STREAM_KEY llm:inference:requests RESULT_KEY_PREFIX llm:result: async def call_tgi_stream(payload: Dict[str, Any]) - AsyncGenerator[str, None]: 异步调用TGI的流式生成接口 async with aiohttp.ClientSession() as session: async with session.post( TGI_SERVER_URL, jsonpayload, headers{Content-Type: application/json} ) as response: async for line in response.content: if line: decoded_line line.decode(utf-8).strip() if decoded_line.startswith(data: ): data json.loads(decoded_line[6:]) token data.get(token, {}).get(text, ) if token: yield token async def process_requests(): 主处理循环从Redis Stream消费并处理请求 last_id 0-0 # 从最开始读取 while True: try: # 阻塞读取新消息最多等待5秒 messages redis_client.xread({STREAM_KEY: last_id}, count10, block5000) if not messages: continue for stream, message_list in messages: for message_id, message_data in message_list: last_id message_id task_id message_data[task_id] user_question message_data[question] session_id message_data[session_id] # 构建TGI请求负载包含历史会话简化处理实际应从缓存读取 payload { inputs: user_question, parameters: { max_new_tokens: 512, temperature: 0.7, stream: True # 启用流式输出 } } full_response result_key f{RESULT_KEY_PREFIX}{task_id} # 流式处理生成结果 async for token in call_tgi_stream(payload): full_response token # 可以在这里将token实时推送到WebSocket或另一个队列实现真正的流式返回给用户 # 此处简化先拼接完整响应 # 将完整响应存入Redis供API层获取 result_data { task_id: task_id, response: full_response, status: completed } redis_client.setex(result_key, 300, json.dumps(result_data)) # 结果保留5分钟 except Exception as e: print(fError processing request: {e}) await asyncio.sleep(1) if __name__ __main__: asyncio.run(process_requests())4. 性能测试数据不说谎优化完成后我们使用locust进行了压测。模拟了100个并发用户持续提问的场景。指标优化前 (朴素API)优化后 (TGI队列)提升幅度QPS~2.5~7.8212%平均延迟4200ms850ms-80%P99延迟12500ms2200ms-82%GPU内存占用14GB (FP16)8.4GB (NF4量化)-40%可以看到吞吐量QPS提升了3倍多延迟大幅下降尤其是P99延迟的改善对用户体验至关重要。内存占用减少让我们有机会在单卡上部署更大的模型或者在原卡上服务更多并发。5. 避坑指南那些我们踩过的“坑”长文本内存溢出用户可能粘贴大段文档咨询。如果不对输入长度做限制极易撑爆KV缓存。解决方案在API层强制截断比如只取最近4096个token。更优雅的做法是使用“滑动窗口”注意力只保留最近的部分上下文。对话状态幂等性网络可能超时用户可能重复提交。必须保证同一会话内相同的问题输入能得到相同的回答并且上下文不会错乱。解决方案为每个会话session_id在Redis中维护一个独立的KV缓存键。每次请求携带上次的缓存IDTGI支持传入past_key_values来延续对话。同时对请求内容生成唯一哈希作为幂等键。监控推理服务健康TGI服务可能因OOM或显存碎片化而崩溃。解决方案使用prometheusgrafana监控服务的GPU利用率、内存占用、请求队列长度和错误率。设置告警并在Worker中实现简单的重试和降级逻辑如失败后返回一个预定义的兜底回答。6. 延伸思考还能更进一步吗目前的优化已经能应对大多数场景但追求极致永无止境模型蒸馏能否用一个小模型如1B参数去学习我们客服场景下大模型7B/13B的行为蒸馏后的模型推理速度会更快成本更低。难点在于如何保持回答的准确性和泛化能力。硬件加速除了GPU是否可以尝试部署到专用的AI推理芯片如AWS Inferentia, Google TPU这些芯片在性价比上可能有惊喜。更智能的调度目前的动态批处理主要看token数量。能否引入优先级、模型预热、请求预测如根据时间预测流量高峰等实现更精细的调度写在最后这套优化方案实施下来最大的感触是优化不是某个“银弹”而是一个系统工程。从模型选择、服务部署、到架构设计、监控运维每一步都需要结合业务场景仔细考量。对于智能客服这种对成本和实时性都敏感的应用在项目早期就引入这些效率优化思维能避免后期很多重构的麻烦。现在我们的客服系统终于能流畅地应对每天的访问高峰了。不过技术总是在发展比如最近流行的MoE混合专家模型在保持效果的同时计算量更少。你是否在LLM效率优化上有其他独特的经验或想法欢迎一起探讨。