LLM服务

LLM缓存的价值 LLM推理是昂贵的——每次调用消耗GPU算力和API费用。但很多请求是重复或高度相似的。缓存可以让"已经算过的不再重算"，是投入产出比最高的优化手段。 三层缓存架构 请求 → L1: 精确缓存 → L2: 语义缓存 → L3: 前缀缓存 → LLM L1：精确缓存 import redis.asyncio as redis class ExactCache: """精确匹配缓存：相同输入返回相同输出""" def __init__(self, redis_url="redis://localhost:6379"): self.redis = redis.from_url(redis_url) def cache_key(self, model, messages, temperature, max_tokens): """生成缓存键""" content = json.dumps({ "model": model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, }, sort_keys=True, ensure_ascii=False) return hashlib.sha256(content.encode()).hexdigest() async def get(self, model, messages, **params): key = self.cache_key(model, messages, params.get("temperature", 0.7), params.get("max_tokens", 2048)) cached = await self.redis.get(key) return json.loads(cached) if cached else None async def set(self, model, messages, response, ttl=3600, **params): key = self.cache_key(model, messages, params.get("temperature", 0.7), params.get("max_tokens", 2048)) await self.redis.setex(key, ttl, json.dumps(response, ensure_ascii=False)) 适用场景 temperature=0的确定性输出 常见问题FAQ 系统提示词相同的请求 L2：语义缓存 class SemanticCache: """语义缓存：相似查询命中缓存""" def __init__(self, vector_store, embed_model, threshold=0.95): self.store = vector_store self.embed_model = embed_model self.threshold = threshold async def get(self, query, model="default", **params): # 向量化查询 query_embedding = await self.embed_model.embed(query) # 搜索相似缓存 results = await self.store.search( query_embedding, filter={"model": model}, top_k=1 ) if results and results[0]["score"] >= self.threshold: cached = json.loads(results[0]["document"]) # 检查参数兼容性 if self.params_compatible(cached["params"], params): return cached["response"] return None async def set(self, query, response, model="default", ttl=3600, **params): embedding = await self.embed_model.embed(query) await self.store.add( id=str(uuid.uuid4()), embedding=embedding, document=json.dumps({ "query": query, "response": response, "model": model, "params": params, "timestamp": time.time(), }, ensure_ascii=False), ttl=ttl ) def params_compatible(self, cached_params, request_params): """检查缓存参数是否兼容当前请求""" # temperature差异较大则不兼容 if abs(cached_params.get("temperature", 0.7) - request_params.get("temperature", 0.7)) > 0.1: return False return True 语义缓存的阈值选择 阈值 命中率 准确率 适用场景 0.99 低 极高 严格场景（医疗、法律） 0.95 中 高 通用场景 0.90 高 中 FAQ类场景 0.85 很高 低 不推荐（风险大） L3：前缀缓存（KV Cache共享） class PrefixCacheManager: """前缀缓存：共享相同前缀的KV Cache""" def __init__(self): self.prefix_cache = {} # prefix_hash -> kv_cache def compute_prefix_hash(self, messages): """计算消息前缀的哈希""" # 系统提示 + 历史对话通常构成共享前缀 prefix_text = json.dumps(messages[:-1]) # 除最后一条消息 return hashlib.sha256(prefix_text.encode()).hexdigest() async def get_kv_cache(self, messages): """获取前缀的KV Cache""" prefix_hash = self.compute_prefix_hash(messages) return self.prefix_cache.get(prefix_hash) async def store_kv_cache(self, messages, kv_cache): """存储前缀的KV Cache""" prefix_hash = self.compute_prefix_hash(messages) self.prefix_cache[prefix_hash] = kv_cache vLLM内置前缀缓存支持： ...

引言 LLM服务框架决定了模型的推理速度、资源利用率和最终的服务成本。2026年，vLLM、TGI、TensorRT-LLM等框架在性能上你追我赶。本文将通过系统化测试，帮你选择最佳的LLM服务框架。 参评框架 框架 厂商 版本 特点 vLLM UC Berkeley 0.8 PagedAttention，通用性强 TGI HuggingFace 3.0 生态丰富，易用 TensorRT-LLM NVIDIA 0.15 NVIDIA官方，性能极致 llama.cpp 开源 b3500 CPU/GPU通用，轻量 MLServer Seldon 1.5 企业级，多协议 Ollama Ollama 0.5 最易用，生态好 LMDeploy 上海AI Lab 0.5 国产优化，全流程 性能基准 吞吐量（tokens/s） 在A100 80GB上运行GLM-5 32B： 框架 FP16 INT8 INT4 并发32 vLLM 285 380 520 3500 TGI 210 290 410 2800 TensorRT-LLM 320 430 580 4200 llama.cpp 85 150 210 - LMDeploy 270 365 500 3200 延迟 单请求延迟（P95）： ...

延迟的两个关键指标 LLM服务的延迟分为两部分： TTFT（Time To First Token）：从请求到第一个token返回的时间 TPOT（Time Per Output Token）：每个后续token的生成时间 用户感知延迟 = TTFT + (输出token数 × TPOT)。优化需要分别针对这两个指标。 TTFT优化 预填充优化 TTFT主要由预填充（处理输入prompt）时间决定： # 1. 分块预填充：避免长prompt阻塞短请求 vllm serve model --enable-chunked-prefill --max-num-batched-tokens 4096 # 2. 前缀缓存：共享系统提示词的KV Cache vllm serve model --enable-prefix-caching # 3. 减少输入长度：精简系统提示词 # 差：500 token的系统提示词 # 好：150 token的精简系统提示词 模型预热 async def warmup_model(model_name): """服务启动时预热模型""" # 预加载模型到GPU dummy_input = "warmup" await llm.generate(dummy_input, max_tokens=1) # 预填充常见前缀 common_prompts = [ "你是一个专业助手", "请根据以下信息回答问题", ] for prompt in common_prompts: await llm.generate(prompt, max_tokens=1) TPOT优化 推测解码 # vLLM启用推测解码 vllm serve model \ --speculative-model /models/draft-model \ --num-speculative-tokens 5 \ --speculative-draft-tensor-parallel-size 1 量化 # INT8量化（2倍加速，精度损失<1%） vllm serve model --quantization awq --dtype float16 # INT4量化（3倍加速，精度损失3-5%） vllm serve model --quantization gptq --dtype float16 KV Cache优化 # FP8 KV Cache（减少显存带宽压力） vllm serve model --kv-cache-dtype fp8 网络层优化 流式响应 # 流式响应让用户更早看到输出 @app.post("/chat") async def chat(): async def stream(): yield "data: " # 立即发送头部 async for token in llm.astream(messages): yield json.dumps({"content": token}) + "\n" return StreamingResponse(stream()) 连接复用 # 使用HTTP/2或WebSocket减少连接建立开销 import httpx # 全局复用客户端 client = httpx.AsyncClient( http2=True, limits=httpx.Limits(max_connections=100, max_keepalive_connections=20), keepalive_expiry=30, ) 调度优化 优先级调度 class PriorityScheduler: """交互请求优先于批处理请求""" async def schedule(self, request): if request.type == "interactive": # 交互请求：立即处理 return await self.process_immediately(request) else: # 批处理请求：低峰期处理 return await self.queue_for_later(request) 请求预取 class RequestPrefetcher: """预测用户可能的下一步请求，提前计算""" async def on_user_typing(self, session_id): """用户正在输入时，预计算可能的请求""" likely_queries = await self.predict_queries(session_id) for query in likely_queries[:2]: # 预计算2个最可能的查询 cache_key = f"prefetch:{session_id}:{query}" if not await self.cache.exists(cache_key): result = await self.llm.generate(query) await self.cache.setex(cache_key, 30, result) 端到端优化清单 优化项 TTFT改善 TPOT改善 实现难度 分块预填充 30-50% — 低 前缀缓存 40-60% — 低 量化 10-20% 50-100% 中 推测解码 — 50-100% 中 流式响应 感知80%↓ — 低 模型分层 20-40% 30-50% 中 监控 # 延迟分解监控 class LatencyBreakdown: def record(self, request_start, first_token, request_end): ttft = first_token - request_start total = request_end - request_start tpot = (total - ttft) / max(output_tokens - 1, 1) metrics = { "ttft_ms": ttft * 1000, "tpot_ms": tpot * 1000, "total_ms": total * 1000, } # 告警 if ttft > 2.0: # TTFT > 2秒 alert("High TTFT") if tpot > 0.05: # TPOT > 50ms alert("High TPOT") 结语 LLM服务延迟优化需要从TTFT和TPOT两个维度系统推进。分块预填充和前缀缓存是最有效的TTFT优化手段，量化和推测解码是TPOT的核心优化手段。流式响应虽然不改变实际延迟，但显著改善用户感知。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

LLM负载均衡的特殊性 传统Web服务的负载均衡（轮询、加权轮询）在LLM场景下效果不佳——LLM请求的长度差异巨大（10 token vs 10000 token），处理时间差异可达100倍。简单的轮询会导致某些节点被长请求占满，而短请求也被迫排队。 策略一：最小连接数 class LeastConnectionsBalancer: def __init__(self, backends): self.backends = {b: 0 for b in backends} # backend -> active_connections self.lock = asyncio.Lock() async def get_backend(self): async with self.lock: backend = min(self.backends, key=self.backends.get) self.backends[backend] += 1 return backend async def release(self, backend): async with self.lock: self.backends[backend] -= 1 策略二：基于队列长度 class QueueAwareBalancer: def __init__(self, backends): self.queues = {b: asyncio.Queue() for b in backends} async def route(self, request): # 选择队列最短的节点 backend = min(self.queues, key=lambda b: self.queues[b].qsize()) await self.queues[backend].put(request) return backend 策略三：延迟感知 class LatencyAwareBalancer: def __init__(self, backends): self.backends = backends self.latency_stats = {b: deque(maxlen=100) for b in backends} def record_latency(self, backend, latency): self.latency_stats[backend].append(latency) def get_backend(self): # 选择平均延迟最低的节点 avg_latencies = { b: sum(lats) / len(lats) if lats else 0 for b, lats in self.latency_stats.items() } return min(avg_latencies, key=avg_latencies.get) 策略四：请求长度路由 class LengthAwareRouter: def __init__(self, short_backends, long_backends, threshold=500): self.short_backends = short_backends # 小模型，处理短请求 self.long_backends = long_backends # 大模型，处理长请求 self.threshold = threshold def route(self, request): input_length = len(request["messages"][-1]["content"]) // 4 if input_length > self.threshold: return self.select_least_loaded(self.long_backends) else: return self.select_least_loaded(self.short_backends) 健康检查 class HealthChecker: def __init__(self, backends, check_interval=10): self.backends = {b: {"healthy": True, "last_check": 0} for b in backends} self.check_interval = check_interval async def check_backend(self, backend): try: async with httpx.AsyncClient() as client: resp = await client.get(f"{backend}/health", timeout=5) return resp.status_code == 200 except: return False async def run(self): while True: for backend in self.backends: healthy = await self.check_backend(backend) self.backends[backend]["healthy"] = healthy if not healthy: logger.warning(f"Backend {backend} unhealthy") await asyncio.sleep(self.check_interval) 结语 LLM负载均衡需要考虑请求长度差异、节点异构性和KV Cache状态。最小连接数+延迟感知的组合策略在大多数场景下表现最佳。配合健康检查和自动故障转移，可以构建高可用的LLM推理服务。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

为什么需要流式响应？ LLM生成一个完整回答可能需要5-30秒。如果等待完整响应再返回，用户体验极差。流式响应（Streaming）让用户看到"逐字打印"的效果，大幅降低感知延迟。 SSE（Server-Sent Events）方案 服务端实现 from fastapi import FastAPI from fastapi.responses import StreamingResponse import json app = FastAPI() @app.post("/chat/stream") async def chat_stream(request: dict): async def event_stream(): # 调用LLM的流式接口 async for chunk in llm.astream( messages=[{"role": "user", "content": request["message"]}] ): data = json.dumps({ "content": chunk.content, "role": "assistant" }, ensure_ascii=False) yield f"data: {data}\n\n" # 发送结束标记 yield f"data: [DONE]\n\n" return StreamingResponse( event_stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", # Nginx不缓冲 } ) 客户端实现 // 浏览器端SSE const eventSource = new EventSource('/chat/stream'); eventSource.onmessage = (event) => { if (event.data === '[DONE]') { eventSource.close(); return; } const data = JSON.parse(event.data); document.getElementById('output').innerHTML += data.content; }; eventSource.onerror = (error) => { console.error('SSE error:', error); eventSource.close(); }; Python客户端 import httpx import json async def stream_chat(url, message): async with httpx.AsyncClient() as client: async with client.stream( "POST", f"{url}/chat/stream", json={"message": message}, timeout=60.0 ) as response: async for line in response.aiter_lines(): if line.startswith("data: "): data = line[6:] if data == "[DONE]": break chunk = json.loads(data) print(chunk["content"], end="", flush=True) WebSocket方案 服务端 from fastapi import FastAPI, WebSocket app = FastAPI() @app.websocket("/ws/chat") async def websocket_chat(ws: WebSocket): await ws.accept() try: while True: message = await ws.receive_text() # 流式生成 async for chunk in llm.astream( messages=[{"role": "user", "content": message}] ): await ws.send_json({ "type": "token", "content": chunk.content }) await ws.send_json({"type": "done"}) except WebSocketDisconnect: print("Client disconnected") 客户端 const ws = new WebSocket('ws://localhost:8000/ws/chat'); ws.onopen = () => { ws.send(JSON.stringify({message: '你好'})); }; ws.onmessage = (event) => { const data = JSON.parse(event.data); if (data.type === 'token') { output.innerHTML += data.content; } else if (data.type === 'done') { console.log('Generation complete'); } }; 高级特性 打字机效果 async def typewriter_stream(text, delay=0.03): """模拟打字机效果的流式输出""" for char in text: yield char await asyncio.sleep(delay) 流式中断 class CancellableStream: def __init__(self): self.cancelled = False def cancel(self): self.cancelled = True async def stream(self, llm, messages): async for chunk in llm.astream(messages): if self.cancelled: break yield chunk 并行流式响应 async def parallel_stream(prompts, llm): """同时流式生成多个响应""" async def stream_one(prompt, index): result = [] async for chunk in llm.astream([{"role": "user", "content": prompt}]): result.append({"index": index, "content": chunk.content}) return result tasks = [stream_one(p, i) for i, p in enumerate(prompts)] results = await asyncio.gather(*tasks) # 交错输出 for i in range(max(len(r) for r in results)): for result in results: if i < len(result): yield result[i] 性能优化 缓冲控制 # Nginx配置：禁用缓冲 location /chat/stream { proxy_pass http://backend; proxy_buffering off; # 关闭代理缓冲 proxy_cache off; # 关闭缓存 chunked_transfer_encoding on; proxy_read_timeout 300s; } Token级vs字符级 # Token级流式（推荐）：每个token一个chunk async for token in llm.astream(messages): yield token # 字符级流式：将token拆分为字符 async for token in llm.astream(messages): for char in token.content: yield char await asyncio.sleep(0.01) # 添加微小延迟 结语 流式响应是LLM应用的标配功能。SSE适合单向流式输出，WebSocket适合需要双向交互的场景。关键点：禁用所有层级的缓冲、正确处理中断、保持连接稳定性。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...