负载均衡

大模型推理服务的负载均衡策略：从轮询到智能调度 大模型推理服务与传统Web服务有本质区别：请求处理时间长（秒级而非毫秒级）、资源消耗大（GPU而非CPU）、请求间差异大（短问答和长文本生成的处理时间可能差100倍）。这些特性使传统负载均衡策略表现不佳，需要设计专用的调度算法。 传统负载均衡的局限 经典的负载均衡算法——轮询、最少连接、加权轮询——在LLM推理服务中面临挑战： 轮询调度：不考虑请求复杂度，一个简单查询和一个2000字长文生成被同等对待。结果是某些Worker被长请求阻塞，而空闲Worker在等待轮询到自己。 最少连接：看起来合理，但"连接数"不能反映真实负载。一个Worker可能只有1个连接但正在处理一个30秒的长请求，而另一个Worker有5个连接但都是短请求。按连接数调度反而会把新请求送到已经过载的Worker。 加权轮询：权重是静态配置的，无法适应请求模式的变化和Worker性能的动态波动。 请求感知调度 我们的核心创新是请求感知调度——根据请求的预期复杂度来路由。 请求复杂度预测 在请求进入调度器时，我们快速预测其处理成本： 输入长度：token数量直接影响prefill时间 输出长度预测：基于Prompt类型和历史数据预测生成长度 模型选择：不同模型的计算量不同 工具调用可能性：可能触发工具调用的请求需要更长处理时间 预测模型是一个轻量的梯度提升树，推理延迟<1ms，预测误差在可接受范围内（±20%）。这个精度足以做出合理的调度决策。 多级队列设计 我们设计了基于优先级的多级队列： P0（交互式）：用户正在等待的实时对话，延迟敏感 P1（半交互式）：Agent工具调用、短文档分析 P2（批处理）：长文档翻译、批量摘要等非实时任务 P3（后台）：模型预热、数据集预处理 调度策略是严格优先级 + 同级内最短预计处理优先（SJF）。SJF策略能最小化平均等待时间，但可能导致长任务饥饿。我们引入了aging机制——等待时间超过阈值后优先级自动提升。 模型路由：异构模型的智能调度 在硅基AGI平台中，我们部署了多个不同规模的模型。小模型（7B级）处理简单请求，大模型（70B+级）处理复杂任务。路由策略： 分类器路由 训练一个轻量分类器，根据请求的复杂度自动路由到合适的模型。分类器特征包括：Prompt长度、问题类型（事实问答vs复杂推理）、是否包含代码、历史对话轮次等。 准确率约85%，对于误分类的情况，小模型处理不了的请求会自动升级到大模型。 级联推理 更激进的策略是级联推理：先用小模型处理，如果置信度低于阈值，再升级到大模型。这种策略在平均情况下节省大量推理成本，但增加了系统复杂度和尾部延迟的不确定性。 实践中的经验：级联推理在事实问答类任务上效果最好（简单问题的置信度判断准确），在创意写作类任务上效果较差（“好"的标准不明确，置信度信号弱）。 GPU层面的优化 负载均衡和GPU层面的调度密切相关。几个关键实践： Continuous Batching 传统批处理需要等一个batch全部完成才能处理下一个，而Continuous Batching在每一步都可以有新请求加入和已完成请求离开。这极大提高了GPU利用率，从传统批处理的40%左右提升到80%+。 KV Cache管理 KV Cache是大模型推理中显存的主要消耗。我们实现了PagedAttention式的分页管理，将KV Cache按固定大小分页，允许不同请求的KV Cache共享物理内存页面。这使得显存利用率从30%提升到75%。 Prefix Caching 很多请求共享相同的系统Prompt前缀。我们将常用前缀的KV Cache缓存，新请求只需计算差异部分。对于系统Prompt很长的场景，这可以减少30-50%的prefill计算。 容错与降级 负载均衡层也是容错的第一道防线。我们的设计： 健康检查：每5秒检查Worker健康状态，连续3次失败标记为不可用 优雅降级：当所有大规模模型Worker不可用时，自动降级到小模型，保持服务可用性 熔断机制：当Worker错误率超过10%时触发熔断，请求路由到其他Worker 限流策略：基于令牌桶的限流，保护系统不被突发流量压垮 监控指标 负载均衡效果需要细粒度监控。关键指标： 指标 目标值 说明 P50延迟 <800ms 中位数响应时间 P95延迟 <3s 95%请求的响应时间 P99延迟 <8s 尾部延迟 GPU利用率 >70% 平均GPU计算利用率 请求丢弃率 <0.1% 因过载被拒的请求比例 Worker负载均衡度 >0.8 最小/最大负载比 负载均衡度是最需要关注的指标——如果10个Worker中9个利用率30%而1个是90%，说明调度出了问题。 ...

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论坛 — 全球首个碳基硅基认知交流平台。 ...

为什么需要AI网关？ 当企业使用多个LLM提供商（OpenAI、Anthropic、本地模型等）时，直接对接各家API会面临：密钥管理分散、无法统一限流、缺乏请求日志、故障切换困难。AI网关统一管理所有LLM请求，提供路由、缓存、限流、监控等基础设施。 核心架构 客户端 → AI网关 → LLM提供商A → LLM提供商B → 本地vLLM 实现方案 统一API接口 from fastapi import FastAPI, Request from pydantic import BaseModel app = FastAPI() class ChatRequest(BaseModel): model: str messages: list temperature: float = 0.7 max_tokens: int = 2048 stream: bool = False # 提供商配置 PROVIDERS = { "openai": {"base_url": "https://api.openai.com/v1", "api_key": "..."}, "anthropic": {"base_url": "https://api.anthropic.com", "api_key": "..."}, "local": {"base_url": "http://localhost:8000/v1", "api_key": "..."}, } # 模型到提供商的路由 MODEL_ROUTING = { "gpt-4": "openai", "claude-3": "anthropic", "qwen3-32b": "local", } @app.post("/v1/chat/completions") async def chat_completions(request: ChatRequest): provider = MODEL_ROUTING.get(request.model, "local") config = PROVIDERS[provider] # 转发请求 async with httpx.AsyncClient() as client: response = await client.post( f"{config['base_url']}/chat/completions", json=request.dict(), headers={"Authorization": f"Bearer {config['api_key']}"}, timeout=120 ) return response.json() 故障切换 class FailoverRouter: def __init__(self, routing_config): self.routing = routing_config # {model: [provider1, provider2, ...]} self.health = {p: True for providers in routing_config.values() for p in providers} async def route(self, model, request): providers = self.routing.get(model, ["local"]) for provider in providers: if not self.health[provider]: continue try: result = await self.call_provider(provider, request) return result except Exception as e: logger.warning(f"Provider {provider} failed: {e}") self.health[provider] = False continue raise ServiceUnavailableError("All providers failed") 请求缓存 import hashlib import redis.asyncio as redis class ResponseCache: def __init__(self, redis_url="redis://localhost:6379"): self.redis = redis.from_url(redis_url) def cache_key(self, model, messages, temperature): content = json.dumps({"model": model, "messages": messages, "temp": temperature}) return hashlib.sha256(content.encode()).hexdigest() async def get(self, model, messages, temperature): key = self.cache_key(model, messages, temperature) cached = await self.redis.get(key) return json.loads(cached) if cached else None async def set(self, model, messages, temperature, response, ttl=3600): key = self.cache_key(model, messages, temperature) await self.redis.setex(key, ttl, json.dumps(response)) 限流 from datetime import datetime, timedelta class RateLimiter: def __init__(self, redis): self.redis = redis async def check(self, user_id, limit=60, window=60): key = f"rate:{user_id}:{datetime.now().strftime('%Y%m%d%H%M')}" current = await self.redis.incr(key) if current == 1: await self.redis.expire(key, window) if current > limit: return False return True 日志与监控 class RequestLogger: def __init__(self): self.logger = structlog.get_logger() async def log(self, request, response, user_id, duration): self.logger.info("llm_request", user_id=user_id, model=request.model, input_tokens=response.get("usage", {}).get("prompt_tokens", 0), output_tokens=response.get("usage", {}).get("completion_tokens", 0), duration_ms=duration * 1000, provider=response.get("provider", "unknown"), status="success" if response.get("choices") else "error" ) 部署配置 Docker Compose version: '3.8' services: gateway: build: . ports: - "8080:8080" environment: - REDIS_URL=redis://redis:6379 depends_on: - redis redis: image: redis:7-alpine ports: - "6379:6379" prometheus: image: prom/prometheus ports: - "9090:9090" grafana: image: grafana/grafana ports: - "3000:3000" 结语 AI网关是LLM生产基础设施的核心组件。统一API、故障切换、缓存、限流和监控这五大功能，让LLM服务具备企业级的可靠性和可观测性。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 路由是Agent系统的"大脑"——它决定了每个请求由哪个模型处理、调用哪些工具、使用什么记忆策略。一个优秀的路由架构可以在不增加硬件资源的前提下，将Agent系统的整体性能提升30-50%，同时显著降低运营成本。 2026年，Agent路由已从简单的规则匹配进化到基于语义理解和强化学习的智能路由，成为Agent系统核心竞争力之一。 路由架构演进 阶段1：固定路由 阶段2：规则路由 阶段3：语义路由 ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 所有请求 │ │ 规则匹配 │ │ 语义理解 │ │ ↓ │ │ ↓ │ │ ↓ │ │ 同一模型 │ │ 多模型分发 │ │ 意图分类 │ │ 同一工具 │ │ 工具选择 │ │ 智能选模型 │ └──────────┘ └──────────┘ └──────────┘ 阶段4：学习型路由 ┌──────────┐ │ 历史数据 │ │ ↓ │ │ RL优化 │ │ ↓ │ │ 动态路由 │ └──────────┘ 规则路由 class RuleBasedRouter: """基于规则的路由器""" def __init__(self): self.rules = [] def add_rule(self, condition: callable, target: dict, priority: int = 0): """添加路由规则""" self.rules.append({ "condition": condition, "target": target, "priority": priority }) self.rules.sort(key=lambda r: r["priority"], reverse=True) async def route(self, request: dict) -> dict: """路由请求""" for rule in self.rules: if rule["condition"](request): return rule["target"] # 默认路由 return {"model": "gpt-4o-mini", "tools": ["search"]} def setup_default_rules(self): """设置默认规则集""" # 代码相关 → 代码模型 self.add_rule( condition=lambda r: "code" in r["input"].lower(), target={"model": "deepseek-coder", "tools": ["code_exec"]}, priority=10 ) # 数学相关 → 数学增强模型 self.add_rule( condition=lambda r: any(w in r["input"].lower() for w in ["计算", "求解", "公式"]), target={"model": "gpt-4o", "tools": ["calculator"]}, priority=8 ) # 简单问候 → 小模型 self.add_rule( condition=lambda r: len(r["input"]) < 20, target={"model": "gpt-4o-mini", "tools": []}, priority=5 ) 语义路由 class SemanticRouter: """基于语义理解的路由器""" def __init__(self, embedding_model, vector_store): self.embedder = embedding_model self.vectors = vector_store self.routes = {} async def register_route( self, name: str, description: str, examples: list, target: dict ): """注册语义路由""" # 为路由描述和示例生成嵌入 texts = [description] + examples embeddings = await self.embedder.embed_batch(texts) self.routes[name] = { "description": description, "examples": examples, "embeddings": embeddings, "target": target } async def route(self, user_input: str) -> dict: """语义路由""" # 生成输入嵌入 input_embedding = await self.embedder.embed(user_input) # 计算与每个路由的相似度 scores = {} for name, route in self.routes.items(): # 使用所有嵌入的最大相似度 similarities = [ self._cosine_similarity(input_embedding, emb) for emb in route["embeddings"] ] scores[name] = max(similarities) # 选择最匹配的路由 best_route = max(scores, key=scores.get) best_score = scores[best_route] # 如果置信度不足，使用默认路由 if best_score < 0.7: logger.info( f"Low confidence route: {best_route} ({best_score:.2f}), " f"using default" ) return self.routes.get("default", {}).get("target", {}) return self.routes[best_route]["target"] @staticmethod def _cosine_similarity(a, b) -> float: return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) 学习型路由 class LearningRouter: """基于历史数据的学习型路由器""" def __init__(self, feature_extractor, model_registry): self.features = feature_extractor self.models = model_registry self.routing_history = [] self.performance_predictor = None # 训练好的ML模型 async def route(self, request: dict) -> dict: """智能路由——预测最佳模型和工具组合""" # 提取请求特征 features = self._extract_features(request) # 预测每个候选模型的性能 candidates = await self._get_candidates(features) predictions = [] for candidate in candidates: perf = await self._predict_performance(features, candidate) predictions.append({ "candidate": candidate, "predicted_quality": perf["quality"], "predicted_latency": perf["latency"], "predicted_cost": perf["cost"], "score": self._compute_score(perf) }) # 选择综合得分最高的 best = max(predictions, key=lambda p: p["score"]) return best["candidate"] def _extract_features(self, request: dict) -> dict: """提取请求特征""" return { "input_length": len(request["input"]), "language": self._detect_language(request["input"]), "complexity_score": self._estimate_complexity(request["input"]), "has_code": "```" in request["input"], "has_math": any(c in request["input"] for c in ["∑", "∫", "√"]), "requires_search": self._needs_search(request["input"]), "conversation_depth": request.get("turn_count", 0), "user_tier": request.get("user_tier", "free"), } def _compute_score(self, perf: dict) -> float: """综合评分""" # 质量权重0.5，延迟权重0.3，成本权重0.2 return ( perf["predicted_quality"] * 0.5 + (1 - min(perf["predicted_latency"] / 5000, 1)) * 0.3 + (1 - min(perf["predicted_cost"] / 0.1, 1)) * 0.2 ) async def record_outcome( self, request: dict, route_decision: dict, outcome: dict ): """记录路由结果，用于持续学习""" self.routing_history.append({ "features": self._extract_features(request), "decision": route_decision, "outcome": outcome, # quality_score, latency, cost "timestamp": datetime.now() }) # 定期重训练 if len(self.routing_history) % 1000 == 0: await self._retrain() 多模型调度 class MultiModelScheduler: """多模型调度器""" def __init__(self): self.models = { "gpt-4o": { "strength": ["reasoning", "code", "creative"], "cost_per_1k": 0.015, "avg_latency_ms": 1500, "context_window": 128000, "max_concurrent": 10, }, "gpt-4o-mini": { "strength": ["simple_qa", "summarization"], "cost_per_1k": 0.0003, "avg_latency_ms": 500, "context_window": 128000, "max_concurrent": 50, }, "claude-3.5-sonnet": { "strength": ["analysis", "writing", "long_context"], "cost_per_1k": 0.008, "avg_latency_ms": 1200, "context_window": 200000, "max_concurrent": 20, }, "deepseek-coder-v2": { "strength": ["code", "debug"], "cost_per_1k": 0.0014, "avg_latency_ms": 800, "context_window": 128000, "max_concurrent": 30, }, } self.model_loads = {m: 0 for m in self.models} async def select_model(self, request: dict) -> str: """选择最优模型""" scores = {} for model_name, config in self.models.items(): # 能力匹配分 capability_score = self._capability_match( request, config["strength"] ) # 负载分（负载越低越好） load_ratio = self.model_loads[model_name] / config["max_concurrent"] load_score = 1 - load_ratio # 成本分 estimated_tokens = len(request["input"]) // 4 cost = estimated_tokens * config["cost_per_1k"] cost_score = 1 - min(cost / 0.5, 1) # 综合得分 scores[model_name] = ( capability_score * 0.5 + load_score * 0.3 + cost_score * 0.2 ) return max(scores, key=scores.get) 级联路由 class CascadingRouter: """级联路由——先试便宜模型，不够再升级""" CASCADE = [ {"model": "gpt-4o-mini", "confidence_threshold": 0.85}, {"model": "gpt-4o", "confidence_threshold": 0.75}, {"model": "gpt-4o", "confidence_threshold": 0.0}, # 最终兜底 ] async def route_with_cascade(self, request: dict) -> dict: """级联路由""" total_cost = 0 for stage in self.CASCADE: model = stage["model"] threshold = stage["confidence_threshold"] response = await self._call_model(model, request) total_cost += response["cost"] if response["confidence"] >= threshold: return { **response, "model_used": model, "total_cost": total_cost, "cascade_stages": 1 } # 将低置信度响应作为上下文传给下一级 request["previous_attempt"] = response return { **response, "model_used": model, "total_cost": total_cost, "cascade_stages": len(self.CASCADE) } 路由监控指标 指标 说明 目标 路由准确率 路由决策被后续验证为正确的比例 >90% 模型利用率 各模型的负载均衡度 偏差<20% 平均路由延迟 路由决策耗时 <50ms 级联触发率 需要升级模型的比例 <15% 成本节省率 相比全用最强模型的节省比例 >40% 总结 Agent路由架构是系统性能和成本的杠杆点。语义路由实现了基于意图的精准分发，学习型路由通过历史数据持续优化决策，多模型调度在能力、负载和成本之间找到最优平衡，级联路由则提供了"先省后花"的成本优化思路。 ...

引言 混合专家（Mixture of Experts, MoE）架构通过将模型参数划分为多个"专家"子网络，每次推理只激活其中一部分，实现了参数规模与计算量的解耦。这一设计使得MoE模型在保持稠密模型性能的同时，大幅降低了推理成本。2026年，MoE已成为主流大模型（Mixtral、DeepSeek-V3、Qwen3-MoE）的核心架构。本文将深入解析MoE的路由机制与负载均衡策略。 MoE基础架构 标准MoE Layer MoE层由 $N$ 个专家网络和1个门控网络（Router/Gating Network）组成： class MoELayer(nn.Module): def __init__(self, d_model, n_experts, n_active, expert_dim): super().__init__() self.n_experts = n_experts self.n_active = n_active # Top-K激活专家数 self.gate = nn.Linear(d_model, n_experts, bias=False) self.experts = nn.ModuleList([ ExpertBlock(d_model, expert_dim) for _ in range(n_experts) ]) def forward(self, x): """ x: [batch_size, seq_len, d_model] """ B, T, D = x.shape # 门控分数 gate_logits = self.gate(x) # [B, T, n_experts] gate_probs = F.softmax(gate_logits, dim=-1) # Top-K选择 topk_probs, topk_indices = gate_probs.topk(self.n_active, dim=-1) # 重新归一化 topk_probs = topk_probs / topk_probs.sum(dim=-1, keepdim=True) # 专家计算（分散式） output = torch.zeros_like(x) for i in range(self.n_active): expert_idx = topk_indices[..., i] # [B, T] expert_weight = topk_probs[..., i].unsqueeze(-1) # [B, T, 1] # 对每个专家ID，批量计算 for eid in range(self.n_experts): mask = (expert_idx == eid) if mask.any(): expert_input = x[mask] expert_output = self.experts[eid](expert_input) output[mask] += expert_weight[mask] * expert_output return output 主流MoE模型对比 模型 总参数 激活参数 专家数 Top-K 特点 Mixtral 8x7B 47B 13B 8 2 首批开源MoE DeepSeek-V2 236B 21B 160 6 细粒度专家 DeepSeek-V3 671B 37B 256 8 共享专家+细粒度 Qwen3-235B 235B 35B 128 8 GQA+MoE Mixtral 8x22B 141B 39B 8 2 大规模MoE 路由机制详解 Top-K路由 标准Top-K路由： ...

MoE（Mixture of Experts）已成为2026年大模型架构的事实标准。从DeepSeek-V3到GPT-5，MoE让模型在保持推理效率的同时实现万亿级参数规模。本文将深入MoE的技术内核。 1. MoE架构基础 1.1 从稠密到稀疏 标准Transformer的FFN层对所有输入执行相同计算。MoE将FFN替换为多个"专家"网络，每个token仅激活少数专家： $$\text{FFN}{MoE}(x) = \sum{i \in \mathcal{I}(x)} g_i(x) \cdot E_i(x)$$ 其中 $\mathcal{I}(x)$ 为路由到token $x$ 的专家索引集合，$g_i(x)$ 为门控权重，$E_i$ 为第 $i$ 个专家。 1.2 参数与计算量的解耦 MoE的关键优势：参数总量与计算量解耦。 模型 总参数 激活参数 FLOPs/token 等效稠密模型 Mixtral 8x7B 47B 13B ~13B ~13B DeepSeek-V3 671B 37B ~37B ~37B GPT-5 (估) 3T 300B ~300B ~300B MoE用3T参数获得了300B稠密模型的效果，但推理仅需300B的计算量。 2. 路由机制详解 2.1 Top-K路由 最基础也最常用的路由策略： class TopKRouter(nn.Module): def __init__(self, d_model, num_experts, top_k=2): super().__init__() self.gate = nn.Linear(d_model, num_experts, bias=False) self.top_k = top_k self.num_experts = num_experts def forward(self, x): # x: [batch * seq_len, d_model] logits = self.gate(x) # [batch * seq_len, num_experts] # Top-K选择 topk_logits, topk_indices = torch.topk(logits, self.top_k, dim=-1) # Softmax归一化（仅对选中的专家） weights = F.softmax(topk_logits, dim=-1) return weights, topk_indices 2.2 Expert Choice路由 反转路由方向：不是token选专家，而是专家选token。 ...

为什么智能体系统需要特殊的负载均衡？ 传统的负载均衡主要处理无状态的 HTTP 请求，关注的是 QPS、延迟和可用性。但 AI 智能体系统的负载特征与传统 Web 服务有本质区别：单个请求的计算开销极大（一次 LLM 推理可能耗时数秒到数十秒）、请求之间可能存在状态依赖（多轮对话上下文）、且不同模型的资源需求差异巨大。 一个典型的 AGI 智能体系统可能同时包含：轻量级的意图分类（小模型，毫秒级响应）、重量级的推理任务（大模型，数十秒）、长周期的 Agent 工作流（多步骤，分钟级甚至小时级）、以及实时向量检索（RAG pipeline）。这些异构负载如果混在一起不加区分地调度，必然导致资源浪费和体验劣化。 智能体负载均衡的核心维度 1. 请求分类与路由 第一步是对进入系统的请求进行智能分类。不同于传统负载均衡器只看 URL 路径，智能体系统需要根据请求的语义特征进行路由： class RequestRouter: def __init__(self): self.routes = { "classification": {"model": "claude-haiku", "timeout": 2000, "pool": "fast"}, "reasoning": {"model": "gpt-4o", "timeout": 30000, "pool": "deep"}, "agent_workflow": {"model": "agent-orchestrator", "timeout": 300000, "pool": "long"}, "embedding": {"model": "text-embedding-3", "timeout": 1000, "pool": "vector"}, } def route(self, request): req_type = self.classify(request) route_config = self.routes[req_type] # 检查对应池的可用容量 if self.pool_available(route_config["pool"]): return self.dispatch(request, route_config) else: # 降级策略 return self.fallback(request, req_type) 关键设计点在于：不同类型的请求进入不同的处理池，彼此隔离，避免慢请求阻塞快请求。这就是所谓的"泳道隔离"模式。 2. 连接池与令牌桶 对于 LLM API 调用，两个核心限制是并发连接数和 token 速率限制。有效的管理需要同时处理这两个维度： 连接池管理：每个模型端点维护独立的连接池，池大小根据模型的并发能力和成本预算动态调整。例如，对于按量计费的 API，可以设置较高的并发上限；对于有 RPM 限制的 API，需要严格控制在限额以内。 令牌桶限流：不仅仅限制请求数量，更要限制 token 消耗速率。一个请求可能消耗 100 token，另一个可能消耗 10000 token。按 token 而非请求数来限流，才能真正做到公平调度： class TokenBucket: def __init__(self, capacity, refill_rate): self.capacity = capacity # 桶容量（最大突发） self.refill_rate = refill_rate # 每秒补充的 token 数 self.tokens = capacity self.last_refill = time.time() def consume(self, tokens_needed): self._refill() if self.tokens >= tokens_needed: self.tokens -= tokens_needed return True return False def _refill(self): now = time.time() elapsed = now - self.last_refill self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate) self.last_refill = now 3. 优先级队列 并非所有请求都同等重要。用户的实时对话请求应该优先于后台的批量处理任务。一个成熟的优先级策略通常包含 3-4 个级别： ...

为什么传统负载均衡不够？ 传统 LB（Nginx round-robin）对 LLM 不适用。原因： 请求耗时差异巨大：生成 10 字 vs 1000 字，耗时差 100 倍 模型异构：不同实例可能运行不同模型（GPT-4 / Llama / Mistral） 显存敏感：并发请求过多导致 OOM 流式响应：SSE 长连接需要特殊处理 架构总览 ┌──────────────────────────┐ │ API Gateway / LB │ │ (Nginx / Traefik / ENI) │ └────────────┬─────────────┘ ▼ ┌──────────────────────────┐ │ Router (模型感知路由) │ │ - 模型映射 │ │ - 负载策略 │ │ - 熔断/降级 │ └────────────┬─────────────┘ ┌────────────┼────────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Model A │ │ Model B │ │ Model C │ │ x3 实例 │ │ x2 实例 │ │ x1 实例 │ │ (GPT-4) │ │(Llama-3)│ │(Mistral)│ └─────────┘ └─────────┘ └─────────┘ 负载均衡策略 1. 加权轮询（Weighted Round Robin） 根据实例的 GPU 数量和模型大小分配权重。 ...