引言

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：水平拆分

将不同负载特征的服务拆分到不同节点：

# K8s部署——按资源特征分节点
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-llm-inference
spec:
  replicas: 3
  selector:
    matchLabels:
      app: agent-llm-inference
  template:
    metadata:
      labels:
        app: agent-llm-inference
    spec:
      nodeSelector:
        hardware: gpu
      containers:
      - name: inference
        image: agent/llm-inference:v2.0
        resources:
          limits:
            nvidia.com/gpu: 1
            memory: 32Gi
          requests:
            nvidia.com/gpu: 1
            memory: 16Gi
        env:
        - name: MODEL_NAME
          value: "gpt-4o-mini"
        - name: MAX_CONCURRENT
          value: "8"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-tool-executor
spec:
  replicas: 10
  selector:
    matchLabels:
      app: agent-tool-executor
  template:
    metadata:
      labels:
        app: agent-tool-executor
    spec:
      nodeSelector:
        hardware: cpu
      containers:
      - name: executor
        image: agent/tool-executor:v2.0
        resources:
          limits:
            cpu: 4
            memory: 8Gi
          requests:
            cpu: 2
            memory: 4Gi

Phase 3：自动伸缩

# HPA——基于自定义指标的自动伸缩
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: agent-tool-executor-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: agent-tool-executor
  minReplicas: 5
  maxReplicas: 50
  metrics:
  - type: Pods
    pods:
      metric:
        name: agent_active_sessions
      target:
        type: AverageValue
        averageValue: "20"  # 每Pod 20个活跃会话
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 30
      policies:
      - type: Percent
        value: 100  # 最多翻倍
        periodSeconds: 60
    scaleDown:
      stabilizationWindowSeconds: 300  # 缩容保守
      policies:
      - type: Percent
        value: 25  # 每次最多缩25%
        periodSeconds: 120

GPU调度与模型并行

GPU资源管理

class GPUScheduler:
    """GPU资源调度器"""
    
    def __init__(self, k8s_client):
        self.k8s = k8s_client
        self.gpu_nodes = {}
    
    async def schedule_inference(
        self,
        model: str,
        batch_size: int
    ) -> dict:
        """调度LLM推理任务"""
        
        # 获取集群GPU资源状态
        gpu_status = await self._get_gpu_status()
        
        # 按GPU利用率排序
        available_gpus = sorted(
            gpu_status,
            key=lambda g: (
                g["utilization"],
                g["memory_used"] / g["memory_total"]
            )
        )
        
        # 选择最优GPU
        for gpu in available_gpus:
            if self._can_fit(gpu, model, batch_size):
                return {
                    "node": gpu["node"],
                    "gpu_id": gpu["gpu_id"],
                    "estimated_latency": self._estimate_latency(gpu, model)
                }
        
        # 没有可用GPU，触发扩容
        await self._trigger_gpu_scale_up()
        raise GPUResourceExhausted("No available GPU, scaling up...")
    
    def _can_fit(self, gpu_status: dict, model: str, batch: int) -> bool:
        """检查GPU是否有足够资源"""
        model_memory = self._get_model_memory(model)
        available = gpu_status["memory_total"] - gpu_status["memory_used"]
        required = model_memory * batch
        # 预留20%的安全余量
        return available >= required * 1.2

模型并行策略

对于超大模型（如70B+），单GPU无法容纳，需要模型并行：

class ModelParallelConfig:
    """模型并行配置"""
    
    @staticmethod
    def get_config(model_size: str, gpu_type: str) -> dict:
        """获取模型并行配置"""
        configs = {
            "7B": {
                "A100-40GB": {"tensor_parallel": 1, "pipeline_parallel": 1},
                "A100-80GB": {"tensor_parallel": 1, "pipeline_parallel": 1},
            },
            "70B": {
                "A100-40GB": {"tensor_parallel": 4, "pipeline_parallel": 1},
                "A100-80GB": {"tensor_parallel": 2, "pipeline_parallel": 1},
            },
            "175B": {
                "A100-40GB": {"tensor_parallel": 8, "pipeline_parallel": 2},
                "A100-80GB": {"tensor_parallel": 4, "pipeline_parallel": 1},
            }
        }
        return configs.get(model_size, {}).get(gpu_type, {})

连接池与请求队列

class AgentConnectionPool:
    """Agent服务连接池"""
    
    def __init__(self, max_connections: int = 100):
        self.max_connections = max_connections
        self.semaphore = asyncio.Semaphore(max_connections)
        self.pools = {}  # service -> connection pool
    
    async def get_connection(self, service: str):
        """获取服务连接"""
        await self.semaphore.acquire()
        try:
            if service not in self.pools:
                self.pools[service] = await self._create_pool(service)
            return await self.pools[service].acquire()
        except Exception:
            self.semaphore.release()
            raise
    
    async def release_connection(self, service: str, conn):
        """释放连接"""
        await self.pools[service].release(conn)
        self.semaphore.release()


class RequestQueue:
    """请求队列——削峰填谷"""
    
    def __init__(self, max_size: int = 1000, timeout: float = 30.0):
        self.queue = asyncio.Queue(maxsize=max_size)
        self.timeout = timeout
        self.processors = []
    
    async def enqueue(self, request: dict) -> str:
        """入队"""
        request_id = str(uuid.uuid4())
        try:
            await asyncio.wait_for(
                self.queue.put({"id": request_id, "data": request}),
                timeout=self.timeout
            )
            return request_id
        except asyncio.TimeoutError:
            raise QueueFullError("Request queue is full")
    
    async def process(self, handler: callable, num_workers: int = 5):
        """启动处理worker"""
        for i in range(num_workers):
            worker = asyncio.create_task(self._worker(f"worker-{i}", handler))
            self.processors.append(worker)
    
    async def _worker(self, name: str, handler: callable):
        """处理worker"""
        while True:
            item = await self.queue.get()
            try:
                await handler(item["data"])
            except Exception as e:
                logger.error(f"{name} error: {e}")
            finally:
                self.queue.task_done()

数据库扩展策略

Agent系统的状态数据需要合适的扩展策略：

┌────────────────────────────────────────────┐
│            读写分离 + 分片策略              │
│                                            │
│  ┌─────────┐     ┌──────────────────┐     │
│  │ Write   │────▶│ Primary (Sharded)│     │
│  │ Traffic │     │ shard-0 shard-1  │     │
│  └─────────┘     └────────┬─────────┘     │
│                           │               │
│                   ┌───────┼───────┐       │
│                   ▼       ▼       ▼       │
│              ┌──────┐┌──────┐┌──────┐    │
│              │Replica││Replica││Replica│   │
│              │  0   ││  1   ││  2   │    │
│              └──────┘└──────┘└──────┘    │
│                   ▲       ▲       ▲       │
│  ┌─────────┐     │       │       │       │
│  │ Read    │─────┴───────┴───────┘       │
│  │ Traffic │                               │
│  └─────────┘                               │
└────────────────────────────────────────────┘

分片策略选择：

分片键适用场景优势劣势
user_id用户数据隔离查询高效热点用户
session_id会话数据天然隔离跨会话查询难
hash(id)均匀分布无热点范围查询无效
timestamp时序数据范围查询高效新分片写入热点

性能基准与容量规划

class CapacityPlanner:
    """容量规划器"""
    
    # 基准数据（基于2026年典型硬件）
    BENCHMARKS = {
        "llm_inference": {
            "A100-80GB": {"qps": 50, "latency_p99_ms": 200, "max_concurrent": 32},
            "H100-80GB": {"qps": 90, "latency_p99_ms": 120, "max_concurrent": 64},
        },
        "tool_execution": {
            "8vCPU-16GB": {"qps": 200, "latency_p99_ms": 50, "max_concurrent": 50},
        },
        "vector_search": {
            "32GB-RAM": {"qps": 500, "latency_p99_ms": 10, "max_concurrent": 100},
        }
    }
    
    def plan(
        self,
        target_rps: int,
        sla_latency_ms: int,
        components: list
    ) -> dict:
        """生成容量规划"""
        plan = {}
        for component in components:
            benchmark = self.BENCHMARKS.get(component["type"], {})
            for hw, perf in benchmark.items():
                needed = max(
                    target_rps / perf["qps"],
                    1
                )
                plan[component["name"]] = {
                    "hardware": hw,
                    "replicas": max(int(needed) + 1, 3),  # 至少3副本
                    "estimated_cost_monthly": self._estimate_cost(hw, int(needed) + 1),
                    "estimated_latency_p99": perf["latency_p99_ms"]
                }
        return plan

总结

Agent系统的可扩展性设计不是简单的"加机器"，而是需要根据不同组件的负载特征制定差异化的扩展策略。GPU密集型的LLM推理、CPU/IO密集型的工具执行、内存密集型的向量检索需要独立扩展。K8s + GPU Operator提供了基础设施层的管理能力，而应用层的连接池、请求队列、分片策略则是提升资源利用率的关键。

核心原则：先优化单机性能，再水平扩展；先拆分服务，再独立伸缩；先基准测试，再容量规划。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。