大模型应用后端底座设计与高并发支撑实践

张

张建站

2026/6/8 4:33:07

10分钟阅读

大模型应用后端底座设计与高并发支撑实践一、场景痛点LLM 落地工程的系统性挑战大语言模型LLM的能力已经得到了广泛认可但在生产环境中部署和运行 LLM 应用却面临着独特的工程挑战。与传统后端服务不同LLM 应用有着独特的资源特性高并发下 GPU 显存成为瓶颈、推理延迟不可预测、Token 消耗成本高昂、上下文窗口限制严格。一个典型的困境是当产品经理要求支持 10000 并发用户时后端团队发现单卡 A100 80GB 的显存只能支持 10-20 的真实并发。简单地堆机器并不能解决问题——需要从模型层到服务层到应用层的系统性优化。本文将深入探讨构建生产级 LLM 应用后端底座的核心技术包括模型服务化、推理优化、高并发架构、成本控制等关键议题。二、底层机制与原理深度剖析2.1 LLM 推理的计算特性和资源消耗flowchart TD A[输入 Prompt] -- B[Tokenization] B -- C[Embedding] C -- D[Transformer Forward Pass] subgraph GPU 计算 D -- E[Attention 计算] E -- F[FFN 计算] F -- G[Layer Norm] G -- E end D -- H[KV Cache] H -- D G -- I[Output Layer] I -- J[De-tokenization] J -- K[输出 Token] style E fill:#ff6b6b style F fill:#ff6b6b style H fill:#b8d4ffLLM 推理分为两个阶段Prefill 阶段首次推理处理完整的输入 Prompt计算并缓存 Key-ValueKV状态。这个阶段是 Compute-bound计算密集耗时与输入长度近似线性相关。Decode 阶段逐 Token 生成基于 KV Cache 逐个生成输出 Token。这个阶段是 Memory-bound内存密集因为每次只处理一个 Token但需要访问整个模型参数。2.2 高并发场景下的资源瓶颈flowchart LR A[100 并发请求] -- B{GPU 资源} B --|A100 80GB| C[30 QPS] B --|A100 40GB| D[15 QPS] B --|V100 32GB| E[10 QPS] B --|RTX 3090| F[5 QPS] G[Batch Size 增加] -- H[显存溢出] G -- I[延迟上升] style B fill:#FFE4B5并发请求的处理方式直接影响 GPU 利用率和吞吐量。常见的批处理策略Static Batching将多个请求组成固定大小的批次处理简单但会造成资源浪费Dynamic Batching运行时动态组合批次需要调度算法Continuous Batching在批次执行过程中动态添加新请求显著提高 GPU 利用率2.3 推理优化技术全景图flowchart TD subgraph 模型层优化 A[模型量化] -- A1[INT8 量化] A -- A2[INT4 量化] A -- A3[GPTQ/AWQ] end subgraph 推理引擎优化 B[vLLM] -- B1[Paged Attention] B -- B2[Continuous Batching] C[TensorRT-LLM] -- C1[Kernel Fusion] C -- C2[Flash Attention] D[TGI] -- D1[Prefix Caching] D -- D2[ speculative Decoding] end subgraph 服务架构优化 E[负载均衡] -- E1[请求路由] E -- E2[熔断降级] F[缓存] -- F1[Prompt Cache] F -- F2[Result Cache] end subgraph 应用层优化 G[Prompt 优化] -- G1[结构化 Prompt] G -- G2[Few-shot] H[异步处理] -- H1[Streaming] H -- H2[WebSocket] end三、生产级代码实现与最佳实践3.1 基于 vLLM 的模型服务化vLLM 是目前最流行的 LLM 推理引擎之一核心是 PagedAttention 算法和 Continuous Batching# vLLM 模型服务启动脚本使用 vLLM 启动 LLM 推理服务 import argparse from vllm import LLM, SamplingParams def parse_args(): parser argparse.ArgumentParser(descriptionvLLM 模型服务) parser.add_argument(--model, typestr, requiredTrue, help模型路径或 HuggingFace 模型ID) parser.add_argument(--tensor-parallel-size, typeint, default1, help张量并行大小) parser.add_argument(--gpu-memory-utilization, typefloat, default0.9, helpGPU 显存利用率) parser.add_argument(--max-model-len, typeint, default8192, help最大模型长度) parser.add_argument(--port, typeint, default8000, help服务端口) parser.add_argument(--dtype, typestr, defaultauto, help数据类型) return parser.parse_args() def main(): args parse_args() # 初始化 LLM 推理引擎 llm LLM( modelargs.model, tensor_parallel_sizeargs.tensor_parallel_size, gpu_memory_utilizationargs.gpu_memory_utilization, max_model_lenargs.max_model_len, dtypeargs.dtype, trust_remote_codeTrue, # vLLM 自动管理 KV Cache block_size16, # PagedAttention 的块大小 ) print(f模型加载完成GPU 数量: {args.tensor_parallel_size}) print(f最大序列长度: {args.max_model_len}) print(fKV Cache 块大小: 16) # 启动 FastAPI 服务 from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional, List import uvicorn app FastAPI(titleLLM Inference API) class InferenceRequest(BaseModel): prompt: str max_tokens: int 256 temperature: float 0.7 top_p: float 0.95 top_k: int 50 stop: Optional[List[str]] None stream: bool False app.post(/v1/completions) async def create_completion(request: InferenceRequest): try: sampling_params SamplingParams( max_tokensrequest.max_tokens, temperaturerequest.temperature, top_prequest.top_p, top_krequest.top_k, stoprequest.stop, ) # vLLM 的核心优化Continuous Batching outputs llm.generate([request.prompt], sampling_params) return { id: cmpl- str(hash(request.prompt))[:8], choices: [{ text: outputs[0].outputs[0].text, finish_reason: stop, }], usage: { prompt_tokens: outputs[0].prompt_token_ids.__len__(), completion_tokens: len(outputs[0].outputs[0].token_ids), total_tokens: outputs[0].prompt_token_ids.__len__() len(outputs[0].outputs[0].token_ids), } } except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/v1/chat/completions) async def create_chat_completion(request: InferenceRequest): # 处理 chat format prompt format_chat_prompt(request.prompt) return await create_completion(InferenceRequest(**{**request.dict(), prompt: prompt})) def format_chat_prompt(messages: List[dict]) - str: # 简化的 chat format return \n.join([f{m[role]}: {m[content]} for m in messages]) # 启动服务 uvicorn.run(app, host0.0.0.0, portargs.port) if __name__ __main__: main()3.2 高并发请求调度器# 智能请求调度器结合优先级队列和自动扩缩容的请求调度器 import asyncio import time from dataclasses import dataclass, field from typing import Optional, List from enum import Enum import heapq from collections import defaultdict class RequestPriority(Enum): HIGH 0 # VIP 用户/付费用户 NORMAL 1 # 普通用户 LOW 2 # 后台任务/批量处理 dataclass(orderTrue) class InferenceRequest: priority: int arrival_time: float field(compareFalse) request_id: str field(compareFalse) prompt: str field(compareFalse) max_tokens: int field(compareFalse) metadata: dict field(compareFalse) future: asyncio.Future field(compareFalse, defaultNone) class LLMScheduler: def __init__( self, max_concurrent: int 10, timeout: float 60.0, enable_adaptive_scaling: bool True ): self.max_concurrent max_concurrent self.timeout timeout self.enable_adaptive_scaling enable_adaptive_scaling # 优先级队列 self.high_priority_queue [] self.normal_priority_queue [] self.low_priority_queue [] # 并发控制 self.active_requests 0 self.request_counter 0 # 统计信息 self.stats { total_requests: 0, completed_requests: 0, failed_requests: 0, avg_latency: 0, } # 自适应扩缩容 self.scale_factor 1.0 self.last_scale_time time.time() async def submit( self, prompt: str, max_tokens: int, priority: RequestPriority RequestPriority.NORMAL, metadata: dict None ) - str: 提交推理请求 request_id freq_{self.request_counter} self.request_counter 1 request InferenceRequest( prioritypriority.value, arrival_timetime.time(), request_idrequest_id, promptprompt, max_tokensmax_tokens, metadatametadata or {}, futureasyncio.Future() ) # 根据优先级加入不同队列 if priority RequestPriority.HIGH: heapq.heappush(self.high_priority_queue, request) elif priority RequestPriority.NORMAL: heapq.heappush(self.normal_priority_queue, request) else: heapq.heappush(self.low_priority_queue, request) self.stats[total_requests] 1 # 触发调度 asyncio.create_task(self._schedule()) return request_id async def _schedule(self): 调度请求到推理引擎 if self.active_requests self.max_concurrent * self.scale_factor: return # 已达上限 # 优先级调度先处理高优先级 request None if self.high_priority_queue: request heapq.heappop(self.high_priority_queue) elif self.normal_priority_queue: request heapq.heappop(self.normal_priority_queue) elif self.low_priority_queue: request heapq.heappop(self.low_priority_queue) if request is None: return self.active_requests 1 try: # 执行推理带超时 result await asyncio.wait_for( self._execute_inference(request), timeoutself.timeout ) request.future.set_result(result) self.stats[completed_requests] 1 except asyncio.TimeoutError: request.future.set_exception( TimeoutError(fRequest {request.request_id} timeout after {self.timeout}s) ) self.stats[failed_requests] 1 except Exception as e: request.future.set_exception(e) self.stats[failed_requests] 1 finally: self.active_requests - 1 # 自适应扩缩容检查 if self.enable_adaptive_scaling: self._check_scaling() async def _execute_inference(self, request: InferenceRequest): 执行实际的推理调用 # 这里调用 vLLM 或其他推理引擎 # 简化实现 await asyncio.sleep(0.1) # 模拟推理时间 return { text: fGenerated text for: {request.prompt[:50]}..., tokens_used: request.max_tokens, latency: 0.1 } def _check_scaling(self): 检查是否需要扩缩容 current_time time.time() # 每分钟检查一次 if current_time - self.last_scale_time 60: return self.last_scale_time current_time # 基于队列长度调整 queue_length ( len(self.high_priority_queue) len(self.normal_priority_queue) len(self.low_priority_queue) ) # 队列过长时扩容 if queue_length self.max_concurrent * 5: self.scale_factor min(2.0, self.scale_factor 0.2) elif queue_length self.max_concurrent: self.scale_factor max(0.5, self.scale_factor - 0.1) def get_stats(self) - dict: 获取调度器统计信息 total self.stats[completed_requests] self.stats[failed_requests] if total 0: success_rate self.stats[completed_requests] / total else: success_rate 1.0 return { **self.stats, success_rate: success_rate, active_requests: self.active_requests, queue_size: ( len(self.high_priority_queue) len(self.normal_priority_queue) len(self.low_priority_queue) ), scale_factor: self.scale_factor, }3.3 Token 消耗追踪与成本控制# Token 消耗追踪系统实时追踪 Token 消耗支持成本分摊和告警 from dataclasses import dataclass from typing import Dict, Optional, List from datetime import datetime, timedelta from collections import defaultdict import asyncio dataclass class TokenUsage: prompt_tokens: int completion_tokens: int total_tokens: int model: str timestamp: datetime cost: float dataclass class UserQuota: daily_limit: int monthly_limit: int current_daily_usage: int 0 current_monthly_usage: int 0 last_reset_date: datetime None class TokenTracker: # 模型定价每 1M tokens 的价格 MODEL_PRICING { gpt-4: {input: 30.0, output: 60.0}, # $/1M tokens gpt-3.5-turbo: {input: 0.5, output: 1.5}, claude-3: {input: 3.0, output: 15.0}, } def __init__(self): # 用户配额配置 self.user_quotas: Dict[str, UserQuota] {} # Token 使用记录 self.usage_records: List[TokenUsage] [] # 告警阈值 self.alert_thresholds { daily_quota_percent: 0.8, # 日配额 80% 时告警 monthly_quota_percent: 0.9, # 月配额 90% 时告警 burst_rate: 10000, # 突发速率告警tokens/分钟 } # 成本分摊记录 self.cost_allocation: Dict[str, Dict[str, float]] defaultdict( lambda: defaultdict(float) ) def calculate_cost( self, prompt_tokens: int, completion_tokens: int, model: str ) - float: 计算 API 调用成本 pricing self.MODEL_PRICING.get(model, {input: 0, output: 0}) input_cost (prompt_tokens / 1_000_000) * pricing[input] output_cost (completion_tokens / 1_000_000) * pricing[output] return input_cost output_cost async def record_usage( self, user_id: str, prompt_tokens: int, completion_tokens: int, model: str, metadata: Optional[dict] None ) - TokenUsage: 记录一次 Token 使用 total_tokens prompt_tokens completion_tokens cost self.calculate_cost(prompt_tokens, completion_tokens, model) usage TokenUsage( prompt_tokensprompt_tokens, completion_tokenscompletion_tokens, total_tokenstotal_tokens, modelmodel, timestampdatetime.now(), costcost ) # 记录到使用历史 self.usage_records.append(usage) # 更新用户配额 self._update_user_quota(user_id, total_tokens) # 分摊成本 if metadata: self._allocate_cost(user_id, metadata, cost) # 检查是否需要告警 await self._check_alerts(user_id) return usage def _update_user_quota(self, user_id: str, tokens: int): 更新用户配额使用量 quota self.user_quotas.get(user_id) if quota is None: return now datetime.now() today now.date() current_month (now.year, now.month) # 检查是否需要重置日配额 if quota.last_reset_date is None or quota.last_reset_date.date() today: quota.current_daily_usage 0 quota.last_reset_date now quota.current_daily_usage tokens quota.current_monthly_usage tokens def _allocate_cost( self, user_id: str, metadata: dict, cost: float ): 成本分摊到项目/部门 project metadata.get(project, default) department metadata.get(department, default) self.cost_allocation[user_id][project] cost def _check_alerts(self, user_id: str) - List[dict]: 检查是否触发告警 quota self.user_quotas.get(user_id) if quota is None: return [] alerts [] # 日配额告警 daily_percent quota.current_daily_usage / quota.daily_limit if daily_percent self.alert_thresholds[daily_quota_percent]: alerts.append({ type: daily_quota_warning, user_id: user_id, usage_percent: daily_percent, message: f日配额使用已达 {daily_percent:.1%} }) # 月配额告警 monthly_percent quota.current_monthly_usage / quota.monthly_limit if monthly_percent self.alert_thresholds[monthly_quota_percent]: alerts.append({ type: monthly_quota_warning, user_id: user_id, usage_percent: monthly_percent, message: f月配额使用已达 {monthly_percent:.1%} }) return alerts def get_user_usage_report( self, user_id: str, days: int 30 ) - dict: 获取用户使用报告 cutoff datetime.now() - timedelta(daysdays) recent_usage [ u for u in self.usage_records if u.timestamp cutoff ] total_prompt sum(u.prompt_tokens for u in recent_usage) total_completion sum(u.completion_tokens for u in recent_usage) total_cost sum(u.cost for u in recent_usage) return { user_id: user_id, period_days: days, total_requests: len(recent_usage), total_prompt_tokens: total_prompt, total_completion_tokens: total_completion, total_cost: total_cost, avg_tokens_per_request: ( (total_prompt total_completion) / len(recent_usage) if recent_usage else 0 ), cost_by_project: dict(self.cost_allocation.get(user_id, {})), }四、边界分析与架构权衡4.1 LLM 服务架构选型方案适用场景优点缺点自托管 vLLM有 GPU 资源成本可控性能好运维复杂OpenAI API快速上线简单易用成本高有合规风险云厂商托管企业用户开箱即用安全灵活性差开源模型量化边缘部署低成本能力受限4.2 高并发优化策略策略效果适用场景Continuous Batching3-5x 吞吐提升通用场景PagedAttention (vLLM)2-3x 吞吐提升高并发场景INT8/INT4 量化2-4x 吞吐提升资源受限场景KV Cache 优化显著降低延迟长上下文场景请求调度优化提升用户体验多用户场景五、总结构建生产级 LLM 应用后端底座需要系统性的工程能力推理引擎选型vLLM、TensorRT-LLM、TGI 等各有优劣并发架构设计Continuous Batching 智能调度成本控制Token 追踪配额管理告警机制监控体系Latency、Throughput、Cost 全方位监控关键成功因素容量规划基于业务峰值进行 GPU 资源规划渐进式优化从基准测试开始逐步优化瓶颈点成本意识在模型能力和成本之间找到平衡可观测性建立完善的监控告警体系LLM 工程化是决定 AI 应用成败的关键需要持续投入和优化。