可扩展性

从单机到分布式：AI Agent集群部署实践 当你的AI Agent从实验室原型走向生产环境，从服务一个用户到服务十万用户，单机部署很快成为瓶颈。本文分享我们在硅基AGI平台中将Agent从单机迁移到分布式集群的完整实践经验和踩坑记录。 单机架构的天花板 单机Agent架构简单直接：一个Python进程加载模型、管理对话状态、执行工具调用。在开发阶段这足够了，但生产环境面临三个硬约束： GPU资源瓶颈：单卡显存有限，大模型推理和工具执行争抢资源 并发限制：GIL加上模型推理的串行性，单机并发上限通常在个位数 单点故障：进程崩溃意味着所有用户中断服务 我们的转折点出现在日活达到500时——单机响应延迟P95从2秒飙升到15秒，用户投诉率激增。 集群架构设计 我们采用了基于消息队列的微服务架构，核心组件包括： 任务调度层 调度器是集群的大脑，负责将用户请求路由到合适的Agent Worker。路由策略经历了三个阶段演进： 轮询调度：最简单但忽略了Worker的异构性 负载感知调度：基于Worker的当前队列长度和GPU利用率路由 能力感知调度：根据任务类型（推理密集型、IO密集型）路由到不同配置的Worker 状态管理层 Agent的状态分为两类：对话状态和执行状态。对话状态包括对话历史、用户画像、偏好设置等，存储在Redis集群中，通过一致性哈希分片。执行状态包括当前任务进度、工具调用结果等临时数据，存储在本地内存并定期快照到持久化存储。 状态同步是分布式Agent最复杂的部分。我们的设计原则是"状态分离"——将需要强一致性的状态和可以最终一致的状态分开处理，避免全局锁。 Agent Worker层 Worker是无状态的推理引擎实例，可以水平扩展。每个Worker包含： 模型加载器：支持模型热加载和版本切换 工具执行沙箱：隔离的工具运行环境 心跳上报模块：定期向调度器报告健康状态 消息队列层 我们使用Kafka作为核心消息总线，所有组件间通信通过消息队列解耦。这带来了两个关键好处：流量削峰（用户请求突增时队列缓冲而非直接拒绝）和故障隔离（单个Worker崩溃不影响其他Worker）。 部署实践中的关键问题 GPU资源调度 在多租户环境中，GPU是最稀缺的资源。我们实现了基于优先级的GPU调度器： 实时交互请求优先级最高，独占GPU 批处理任务优先级中等，使用GPU空闲时段 模型加载和预热优先级最低，利用夜间低谷 同时使用vLLM的PagedAttention技术，通过分页管理KV Cache，将GPU利用率从35%提升到78%。 容错与恢复 分布式系统中故障是常态。我们实现了多层容错机制： Worker级：心跳超时30秒自动重启Worker 任务级：执行中断的任务自动重新调度到健康Worker 会话级：对话状态定期快照，故障恢复后从最近快照继续 集群级：跨可用区部署，单AZ故障自动切换 最难处理的是"部分故障"——Worker活着但响应异常缓慢。我们引入了基于超时的熔断机制和基于响应质量评分的异常检测。 版本管理与灰度发布 Agent的版本更新比传统软件复杂得多——不仅是代码变更，还涉及模型权重和Prompt的变更。我们的灰度发布策略： 新版本先在10%流量上运行 对比新旧版本的关键指标（任务完成率、用户满意度、延迟） 自动回滚机制：如果关键指标下降超过阈值，5分钟内自动回滚 灰度期间保留旧版本Worker，确保可以即时切换 监控与可观测性 分布式Agent的可观测性需要关注传统指标（CPU、内存、延迟）和AI特有指标（推理质量、工具调用成功率、上下文窗口利用率）。我们使用Prometheus + Grafana做基础设施监控，自研了AI指标面板用于追踪： 每分钟推理请求数和完成率 各工具调用的成功率和平均耗时 对话轮次分布和上下文长度分布 模型输出的安全审核通过率 成本优化 分布式部署的成本可能快速失控。我们的经验是"先测量再优化"：通过细粒度的成本追踪系统，精确计算每个用户、每个任务的实际资源消耗，然后针对性地优化。 最大的优化收益来自三个方向： 模型蒸馏：用小模型处理简单请求，大模型只处理复杂任务，降低60%推理成本 请求批处理：将多个用户的请求合并为一个batch，提高GPU利用率 弹性伸缩：基于流量预测自动调整Worker数量，闲时缩减到最小规模 结语 从单机到分布式的迁移不是简单的技术升级，而是工程范式的转变。设计阶段多花一天思考，运行阶段少花一周调试。在AI Agent基础设施这个领域，架构的弹性和可观测性比原始性能更重要——因为Agent的行为本就不确定，你需要在不确定的基础上构建可靠的系统。 本文同步发布于 硅基AGI论坛

引言 容量规划是Agent系统运维中最容易被忽视却最关键的环节。一个容量规划不足的系统会在流量高峰时崩溃，而过度规划则会导致巨大的资源浪费。2026年，随着Agent系统规模的扩大，容量规划已从"拍脑袋估算"进化为基于数据驱动的科学决策过程。 容量规划流程 ┌──────────────────────────────────────────────────────────┐ │ 容量规划流程 │ │ │ │ Step 1: 需求预测 │ │ ┌────────────────────────────────────────────┐ │ │ │ 历史数据分析 → 增长趋势 → 流量预测 │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ Step 2: 压测验证 │ │ ┌────────────────▼───────────────────────────┐ │ │ │ 设计压测场景 → 执行压测 → 收集性能数据 │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ Step 3: 资源建模 │ │ ┌────────────────▼───────────────────────────┐ │ │ │ 建立资源消耗模型 → 计算所需资源 │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ Step 4: 容量决策 │ │ ┌────────────────▼───────────────────────────┐ │ │ │ 成本分析 → 容量方案 → 采购/扩容决策 │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ Step 5: 持续监控与调整 │ │ ┌────────────────▼───────────────────────────┐ │ │ │ 实时监控 → 对比预测 → 调整容量计划 │ │ │ └────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────────┘ 需求预测 import numpy as np from sklearn.linear_model import LinearRegression class DemandForecaster: """需求预测器""" def __init__(self, historical_data: list): self.data = historical_data # [{"date": ..., "qps": ..., "sessions": ...}] def forecast( self, horizon_days: int = 30, confidence_interval: float = 0.95 ) -> dict: """预测未来需求""" # 准备训练数据 X = np.array(range(len(self.data))).reshape(-1, 1) y_qps = np.array([d["qps"] for d in self.data]) y_sessions = np.array([d["sessions"] for d in self.data]) # 训练模型 model_qps = LinearRegression() model_qps.fit(X, y_qps) model_sessions = LinearRegression() model_sessions.fit(X, y_sessions) # 预测 future_X = np.array(range( len(self.data), len(self.data) + horizon_days )).reshape(-1, 1) predicted_qps = model_qps.predict(future_X) predicted_sessions = model_sessions.predict(future_X) # 计算置信区间（简化版） residuals_qps = y_qps - model_qps.predict(X) std_qps = np.std(residuals_qps) z_score = 1.96 if confidence_interval == 0.95 else 1.645 forecast = { "horizon_days": horizon_days, "predicted_qps": predicted_qps.tolist(), "predicted_sessions": predicted_sessions.tolist(), "confidence_interval": { "lower_qps": (predicted_qps - z_score * std_qps).tolist(), "upper_qps": (predicted_qps + z_score * std_qps).tolist(), }, "peak_qps": float(np.max(predicted_qps)), "peak_sessions": float(np.max(predicted_sessions)), } return forecast def forecast_with_seasonality(self, horizon_days: int) -> dict: """考虑季节性的预测（如工作日vs周末）""" # 提取季节性模式 daily_pattern = self._extract_daily_pattern() weekly_pattern = self._extract_weekly_pattern() base_forecast = self.forecast(horizon_days, 0.95) # 应用季节性调整 adjusted = [] for i, qps in enumerate(base_forecast["predicted_qps"]): day_of_week = (len(self.data) + i) % 7 hour_of_day = (len(self.data) + i) % 24 seasonal_factor = ( daily_pattern[hour_of_day] * weekly_pattern[day_of_week] ) adjusted.append(qps * seasonal_factor) base_forecast["predicted_qps_seasonal"] = adjusted return base_forecast 压测方案设计 class CapacityTestPlan: """容量测试方案""" TEST_SCENARIOS = [ { "name": "steady_load", "description": "稳态负载测试", "qps": 100, "duration_minutes": 60, "concurrent_users": 500, }, { "name": "burst_load", "description": "突发负载测试", "qps": 500, "duration_minutes": 10, "concurrent_users": 2000, }, { "name": "ramp_up", "description": "逐步加压测试", "start_qps": 50, "end_qps": 1000, "step_qps": 50, "step_duration_minutes": 5, }, { "name": "spike_test", "description": "尖峰测试", "pattern": "spike", # 快速上升到峰值然后下降 "peak_qps": 2000, "spike_duration_minutes": 5, }, { "name": "soak_test", "description": "浸泡测试（长时间运行）", "qps": 200, "duration_hours": 24, }, ] async def run_capacity_tests(self) -> dict: """执行容量测试套件""" results = {} for scenario in self.TEST_SCENARIOS: logger.info(f"Running scenario: {scenario['name']}") result = await self._execute_test_scenario(scenario) results[scenario["name"]] = result # 如果系统已达极限，停止后续测试 if result["status"] == "system_overloaded": logger.warning(f"System overloaded at {scenario['name']}") break return self._analyze_capacity_results(results) def _analyze_capacity_results(self, results: dict) -> dict: """分析容量测试结果""" analysis = { "max_sustainable_qps": 0, "max_concurrent_users": 0, "bottleneck": None, "resource_utilization_at_max": {}, "recommendations": [], } for scenario_name, result in results.items(): if result["error_rate"] < 0.01: # 错误率<1%视为可持续 analysis["max_sustainable_qps"] = max( analysis["max_sustainable_qps"], result["achieved_qps"] ) analysis["max_concurrent_users"] = max( analysis["max_concurrent_users"], result["concurrent_users"] ) # 记录资源利用率 if result["achieved_qps"] > analysis["max_sustainable_qps"] * 0.9: analysis["resource_utilization_at_max"] = result["resource_utilization"] # 识别瓶颈 util = analysis["resource_utilization_at_max"] if util.get("gpu_utilization", 0) > 0.9: analysis["bottleneck"] = "GPU" analysis["recommendations"].append("Add more GPU nodes") elif util.get("cpu_utilization", 0) > 0.9: analysis["bottleneck"] = "CPU" analysis["recommendations"].append("Add more CPU nodes or optimize code") elif util.get("memory_utilization", 0) > 0.85: analysis["bottleneck"] = "Memory" analysis["recommendations"].append("Increase memory or optimize memory usage") return analysis 资源估算模型 class ResourceEstimator: """资源估算器""" # 基于压测数据的资源消耗基准 BENCHMARKS = { "requests_per_gpu": 50, # 每块GPU每秒处理的请求数 "requests_per_cpu": 10, # 每vCPU每秒处理的请求数（非LLM部分） "memory_per_session_mb": 10, # 每会话内存消耗 "storage_per_user_mb": 100, # 每用户存储消耗 } def estimate_resources( self, predicted_qps: float, predicted_sessions: int, growth_margin: float = 0.3 # 30%增长余量 ) -> dict: """估算所需资源""" # 1. GPU资源（LLM推理） required_qps_with_margin = predicted_qps * (1 + growth_margin) gpu_count = int(np.ceil( required_qps_with_margin / self.BENCHMARKS["requests_per_gpu"] )) # 2. CPU资源（路由、工具执行等） cpu_vcpus = int(np.ceil( required_qps_with_margin / self.BENCHMARKS["requests_per_cpu"] )) # 3. 内存资源 memory_gb = int(np.ceil( (predicted_sessions * self.BENCHMARKS["memory_per_session_mb"]) / 1024 )) + 8 # +8GB系统开销 # 4. 存储资源 storage_tb = (predicted_sessions * self.BENCHMARKS["storage_per_user_mb"]) / (1024 * 1024) estimate = { "compute": { "gpu": { "type": "A100-80GB", "count": gpu_count, "utilization_target": 0.75, # 目标利用率75% }, "cpu": { "vcpus": cpu_vcpus, "type": "8vCPU-16GB", "nodes": int(np.ceil(cpu_vcpus / 8)), } }, "memory": { "total_gb": memory_gb, "per_node_gb": 64, "nodes": int(np.ceil(memory_gb / 64)), }, "storage": { "total_tb": storage_tb * 1.5, # 1.5x用于复制和增长 "type": "SSD", }, "network": { "bandwidth_gbps": 10, "load_balancers": int(np.ceil(gpu_count / 8)), } } return estimate def estimate_cost( self, resources: dict, cloud_provider: str = "aws" ) -> dict: """估算成本""" PRICING = { "aws": { "a100_gpu_hour": 4.10, # A100每小时 "ec2_8vcpu_hour": 0.40, # 8vCPU实例 "memory_gb_month": 0.005, # $/GB/月 "storage_tb_month": 50, # $/TB/月 "network_gb": 0.09, # $/GB流量 }, "azure": { "a100_gpu_hour": 3.80, "vm_8vcpu_hour": 0.35, "memory_gb_month": 0.004, "storage_tb_month": 45, "network_gb": 0.08, } } pricing = PRICING[cloud_provider] # 计算月成本 gpu_cost = resources["compute"]["gpu"]["count"] * pricing["a100_gpu_hour"] * 24 * 30 cpu_cost = resources["compute"]["cpu"]["nodes"] * pricing["ec2_8vcpu_hour"] * 24 * 30 memory_cost = resources["memory"]["total_gb"] * pricing["memory_gb_month"] * 30 storage_cost = resources["storage"]["total_tb"] * pricing["storage_tb_month"] total_monthly = gpu_cost + cpu_cost + memory_cost + storage_cost return { "cloud_provider": cloud_provider, "resources": resources, "cost_breakdown": { "gpu": gpu_cost, "cpu": cpu_cost, "memory": memory_cost, "storage": storage_cost, }, "total_monthly_usd": total_monthly, "total_annual_usd": total_monthly * 12, "cost_per_request_usd": total_monthly / (resources["compute"]["gpu"]["count"] * self.BENCHMARKS["requests_per_gpu"] * 24 * 30), } 容量规划报告 class CapacityReport: """容量规划报告生成器""" def generate( self, forecast: dict, capacity_test: dict, resource_estimate: dict, cost_estimate: dict ) -> str: """生成容量规划报告""" report = f""" # Agent系统容量规划报告 **生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M')} --- ## 1. 需求预测 ### 未来{forecast["horizon_days"]}天预测 - **峰值QPS**: {forecast["peak_qps"]:.1f} - **峰值并发会话**: {forecast["peak_sessions"]:.0f} ### 预测曲线 QPS ^ | * | * | * | * | * |*_______ +————————> 时间 (天) 0 {forecast[“horizon_days”]//4} {forecast[“horizon_days”]//2} {forecast[“horizon_days”]*3//4} {forecast[“horizon_days”]} ...

引言 Agent系统的扩展性挑战与传统Web应用截然不同。LLM推理是GPU密集型操作，工具执行可能是CPU或IO密集型，而向量检索则是内存密集型。这意味着简单的"加机器"策略无法有效解决Agent系统的扩展问题。 2026年，K8s + GPU Operator已成为Agent系统部署的事实标准，但如何高效利用集群资源仍然是工程团队面临的核心挑战。 扩展维度分析 Agent系统需要在多个维度上独立扩展： ┌─────────────────────────────────────────────────────────┐ │ Agent系统扩展维度 │ ├─────────────┬──────────────┬──────────────┬────────────┤ │ 并发会话数 │ 推理吞吐量 │ 工具执行并发 │ 记忆检索延迟│ │ (CPU/Mem) │ (GPU) │ (CPU/IO) │ (RAM/SSD) │ ├─────────────┼──────────────┼──────────────┼────────────┤ │ 水平扩展 │ GPU水平扩展 │ 水平扩展 │ 分片+副本 │ │ +Stateless │ +模型并行 │ +无状态 │ +读副本 │ └─────────────┴──────────────┴──────────────┴────────────┘ 从单机到集群的演进路径 Phase 1：单机优化 在扩展之前，先榨干单机性能： import asyncio import uvicorn from concurrent.futures import ThreadPoolExecutor class SingleNodeAgent: """单机Agent——最大化单节点利用率""" def __init__(self): # CPU密集型任务（工具执行） self.cpu_pool = ThreadPoolExecutor( max_workers=8, thread_name_prefix="tool-exec" ) # IO密集型任务（网络请求） self.io_pool = ThreadPoolExecutor( max_workers=32, thread_name_prefix="io-op" ) # LLM推理使用GPU，通过信号量控制并发 self.llm_semaphore = asyncio.Semaphore(4) async def process_request(self, user_input: str) -> str: # 并行执行独立任务 memory_task = asyncio.create_task(self._retrieve_memory(user_input)) tool_task = asyncio.create_task(self._execute_tools(user_input)) memory = await memory_task tool_results = await tool_task # LLM推理（GPU受限） async with self.llm_semaphore: response = await self._llm_inference(user_input, memory, tool_results) return response Phase 2：水平拆分 将不同负载特征的服务拆分到不同节点： ...

扩展性挑战：为什么单机 Agent 走不远 单个 Agent 进程能处理的并发请求受限于 CPU、内存和网络连接数。一个典型的 LLM Agent 单次推理消耗 2-8GB 显存，单卡 GPU 同时只能服务少数几个请求。当 QPS 从个位数涨到百级、千级时，单机方案必然崩溃。 扩展 Agent 系统的核心挑战不是简单的"加机器"，而是： 挑战 描述 影响 状态管理 Agent 会话上下文在哪台机器？ 直接影响会话连续性 GPU 资源 LLM 推理是计算密集型 扩展瓶颈在 GPU 而非 CPU 长连接 流式输出需要持久连接 连接亲和性限制负载均衡 一致性 多 Agent 实例的内存状态同步 分布式共识开销 尾延迟 少数慢请求拖垮整体 P99 需要超时和降级机制 水平扩展策略 1. 无状态 Agent：将状态外部化 扩展的第一原则：Agent 进程本身不持有任何会话状态。所有状态存储在外部系统。 from dataclasses import dataclass, field from typing import Optional, Any import json # ❌ 错误做法：状态在内存中 class StatefulAgent: def __init__(self): self.conversations: dict[str, list] = {} # 内存中的会话状态 def chat(self, session_id: str, message: str) -> str: history = self.conversations[session_id] # 只在本机有效 history.append({"role": "user", "content": message}) # ...处理... return "response" # ✅ 正确做法：状态外部化 class StatelessAgent: """无状态 Agent，所有会话状态存入 Redis""" def __init__(self, state_store: 'SessionStore'): self.store = state_store def chat(self, session_id: str, message: str) -> str: # 每次从外部存储加载会话状态 history = self.store.get_history(session_id) history.append({"role": "user", "content": message}) # 处理逻辑 response = self._generate(history) history.append({"role": "assistant", "content": response}) # 持久化回外部存储 self.store.save_history(session_id, history) return response class SessionStore: """基于 Redis 的会话存储""" def __init__(self, redis_client): self.redis = redis_client self.ttl = 3600 # 会话过期时间 1 小时 def get_history(self, session_id: str) -> list[dict]: data = self.redis.get(f"session:{session_id}") return json.loads(data) if data else [] def save_history(self, session_id: str, history: list[dict]): self.redis.setex( f"session:{session_id}", self.ttl, json.dumps(history, ensure_ascii=False) ) def extend_ttl(self, session_id: str): self.redis.expire(f"session:{session_id}", self.ttl) 2. 分层架构：接入层 + 推理层 + 知识层 """ 三层架构设计： - 接入层：处理 HTTP/WebSocket 连接，会话路由 - 推理层：LLM 推理，可水平扩展 - 知识层：向量数据库、知识图谱 """ from abc import ABC, abstractmethod class GatewayLayer: """接入层：路由请求到推理节点""" def __init__(self, inference_pool: 'InferencePool'): self.pool = inference_pool async def handle_request(self, session_id: str, message: str): # 根据会话亲和性选择节点 node = self.pool.select_node(session_id) return await node.process(session_id, message) class InferenceNode: """推理层：单个推理节点""" def __init__(self, node_id: str, model_path: str): self.node_id = node_id self.model = self._load_model(model_path) self.active_sessions = 0 self.max_sessions = 10 async def process(self, session_id: str, message: str) -> str: self.active_sessions += 1 try: # 推理逻辑 result = await self._infer(message) return result finally: self.active_sessions -= 1 @property def available(self) -> bool: return self.active_sessions < self.max_sessions class InferencePool: """推理节点池""" def __init__(self): self.nodes: list[InferenceNode] = [] def add_node(self, node: InferenceNode): self.nodes.append(node) def select_node(self, session_id: str = None) -> InferenceNode: # 优先选择可用且负载最低的节点 available = [n for n in self.nodes if n.available] if not available: raise RuntimeError("No available inference nodes") return min(available, key=lambda n: n.active_sessions) 3. 推理与工具调用分离 LLM 推理和工具执行有不同的资源特征：推理需要 GPU，工具调用主要消耗 CPU 和网络 I/O。将它们分离到不同服务。 ...