K8s 上的 Agent 弹性部署：从单实例到多副本的自动伸缩实战
一、AI Agent 部署的弹性困境：闲时浪费，忙时排队
把 AI Agent 部署到 K8s 不难，难的是让它既不浪费资源又能扛住突发流量。Agent 的工作模式跟普通 Web 服务完全不同——它不是请求-响应模型，而是长时间运行的有状态任务。一个 Agent 可能跑 30 秒处理简单查询，也可能跑 10 分钟完成复杂的多步骤推理。
这意味着传统的 HPA 策略（基于 CPU/内存利用率）对 Agent 几乎失效。CPU 指标反映不了 Agent 的真实负载——一个 Agent Pod 可能 CPU 只有 30% 但已经排队 50 个任务，也可能 CPU 90% 但只处理一个任务。你需要的是基于任务队列深度的弹性策略。
更棘手的是 Agent 的冷启动问题。LLM 模型加载到 GPU 需要 10-30 秒，如果 HPA 扩容太慢，用户请求会堆积；扩容太快，GPU 资源浪费严重。在云原生环境下，这个问题被放大——GPU 节点的启动时间比 CPU 节点长得多。
别整虚的，直接上方案。这篇文章从 Agent 的实际工作模式出发，给出 K8s 弹性部署的完整方案。
二、Agent 弹性部署架构
flowchart TD
A[用户请求] –> B[API Gateway]
B –> C[任务队列<br/>Redis Stream]
C –> D[Agent Worker Pool]
D –> D1[Agent Pod 1<br/>GPU: T4]
D –> D2[Agent Pod 2<br/>GPU: T4]
D –> D3[Agent Pod N<br/>GPU: A100]
C –> E[队列深度监控]
E –> F[Custom Metrics API]
F –> G[HPA Controller]
G –>|队列深度 > 阈值| H[扩容 Agent Pod]
G –>|队列深度 < 阈值| I[缩容 Agent Pod]
subgraph 资源调度
H –> J[GPU 节点池]
J –> J1[Spot 实例<br/>低成本]
J –> J2[按需实例<br/>稳定兜底]
end
subgraph 优雅上下线
K[Pod PreStop Hook] –> L[完成当前任务]
L –> M[从队列取消注册]
M –> N[安全退出]
end
核心设计思路：任务队列解耦 + 自定义指标驱动 + GPU 节点池分层。
请求不直接打到 Agent Pod，而是进入任务队列。Agent Worker 从队列拉取任务处理。HPA 根据队列深度（而非 CPU）决定扩缩容。GPU 节点池分两层：Spot 实例处理常规负载（成本低），按需实例兜底峰值（稳定可靠）。
三、生产级实现
3.1 Agent Worker：从队列消费任务
# agent_worker.py – Agent Worker 核心逻辑
import asyncio
import signal
import json
import os
from redis import asyncio as aioredis
class AgentWorker:
def __init__(self):
self.redis_url = os.getenv("REDIS_URL", "redis://redis:6379")
self.queue_name = "agent:tasks"
self.consumer_group = "agent-workers"
self.consumer_name = os.getenv("HOSTNAME", "worker-0")
self.running = True
self.current_task = None
async def start(self):
self.redis = aioredis.from_url(self.redis_url)
# 注册优雅退出
loop = asyncio.get_event_loop()
for sig in (signal.SIGTERM, signal.SIGINT):
loop.add_signal_handler(sig, self._graceful_shutdown)
# 确保消费者组存在
try:
await self.redis.xgroup_create(
self.queue_name, self.consumer_group, id="0", mkstream=True
)
except Exception:
pass # 组已存在
print(f"Worker {self.consumer_name} started")
await self._consume_loop()
async def _consume_loop(self):
while self.running:
# 从队列拉取任务，阻塞等待 2 秒
messages = await self.redis.xreadgroup(
self.consumer_group,
self.consumer_name,
{self.queue_name: ">"},
count=1,
block=2000
)
if not messages:
continue
for stream, msgs in messages:
for msg_id, data in msgs:
self.current_task = msg_id
try:
result = await self._process_task(data)
await self._report_result(msg_id, result)
# ACK 任务
await self.redis.xack(
self.queue_name, self.consumer_group, msg_id
)
except Exception as e:
print(f"Task {msg_id} failed: {e}")
# 任务失败，重新入队
await self.redis.xadd(
f"{self.queue_name}:retry",
data
)
await self.redis.xack(
self.queue_name, self.consumer_group, msg_id
)
finally:
self.current_task = None
async def _process_task(self, data: dict) -> dict:
"""执行 Agent 任务"""
task_type = data.get(b"type", b"").decode()
prompt = data.get(b"prompt", b"").decode()
# 调用 LLM 执行 Agent 逻辑
result = await self._run_agent(task_type, prompt)
return {"status": "completed", "result": result}
async def _run_agent(self, task_type: str, prompt: str) -> str:
"""Agent 核心推理逻辑"""
# 这里对接实际的 LLM 推理服务
import httpx
async with httpx.AsyncClient() as client:
resp = await client.post(
"http://llm-service:8000/v1/chat/completions",
json={
"model": "agent-model",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 2048
},
timeout=120
)
return resp.json()["choices"][0]["message"]["content"]
async def _report_result(self, task_id: str, result: dict):
"""上报任务结果"""
await self.redis.publish(
"agent:results",
json.dumps({"task_id": task_id, **result})
)
def _graceful_shutdown(self):
"""优雅退出：完成当前任务后再退出"""
print(f"Worker {self.consumer_name} shutting down gracefully…")
self.running = False
if self.current_task:
print(f"Waiting for task {self.current_task} to complete…")
3.2 自定义指标：暴露队列深度给 HPA
# metrics_exporter.py – 自定义指标导出
from prometheus_client import Gauge, start_http_server
import redis
import time
class QueueMetricsExporter:
def __init__(self):
self.redis_client = redis.from_url("redis://redis:6379")
self.queue_name = "agent:tasks"
# Prometheus 自定义指标
self.queue_depth = Gauge(
"agent_queue_depth",
"Number of pending tasks in agent queue"
)
self.active_workers = Gauge(
"agent_active_workers",
"Number of active agent workers"
)
self.avg_task_duration = Gauge(
"agent_avg_task_duration_seconds",
"Average task processing duration"
)
def run(self, port=9090):
start_http_server(port)
print(f"Metrics exporter started on port {port}")
while True:
self._update_metrics()
time.sleep(5)
def _update_metrics(self):
# 队列深度 = XLEN – 已 ACK 的消息数
try:
pending = self.redis_client.xpending(
self.queue_name, "agent-workers"
)
depth = pending.get("pending", 0) if pending else 0
self.queue_depth.set(depth)
except Exception:
self.queue_depth.set(0)
# 活跃 Worker 数
try:
consumers = self.redis_client.xinfo_consumers(
self.queue_name, "agent-workers"
)
self.active_workers.set(len(consumers))
except Exception:
self.active_workers.set(0)
3.3 K8s 部署清单：HPA + Pod 生命周期管理
# agent-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: agent-worker
labels:
app: agent-worker
spec:
replicas: 2
selector:
matchLabels:
app: agent-worker
template:
metadata:
labels:
app: agent-worker
annotations:
prometheus.io/scrape: "true"
prometheus.io/port: "9090"
spec:
terminationGracePeriodSeconds: 120 # 给 Agent 足够时间完成当前任务
containers:
– name: agent
image: agent-worker:latest
resources:
requests:
nvidia.com/gpu: 1
memory: "4Gi"
cpu: "2"
limits:
nvidia.com/gpu: 1
memory: "8Gi"
cpu: "4"
env:
– name: REDIS_URL
value: "redis://redis:6379"
lifecycle:
preStop:
exec:
command: ["/bin/sh", "-c", "kill -SIGTERM 1"]
readinessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 30
periodSeconds: 10
livenessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 60
periodSeconds: 30
—
# 基于自定义指标的 HPA
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: agent-worker-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: agent-worker
minReplicas: 2
maxReplicas: 20
metrics:
– type: Pods
pods:
metric:
name: agent_queue_depth
target:
type: AverageValue
averageValue: "5" # 每个 Pod 处理 5 个任务为理想状态
behavior:
scaleUp:
stabilizationWindowSeconds: 30
policies:
– type: Pods
value: 4
periodSeconds: 60
scaleDown:
stabilizationWindowSeconds: 300 # 缩容保守，避免抖动
policies:
– type: Pods
value: 1
periodSeconds: 120
3.4 GPU 节点池分层：Spot + 按需
# gpu-nodepool-spot.yaml – Spot 实例节点池
apiVersion: apps/v1
kind: Deployment
metadata:
name: agent-worker-spot
labels:
app: agent-worker
pool: spot
spec:
replicas: 3
template:
spec:
nodeSelector:
node-pool: gpu-spot
tolerations:
– key: "spot-instance"
operator: "Equal"
value: "true"
effect: "NoSchedule"
containers:
– name: agent
image: agent-worker:latest
resources:
requests:
nvidia.com/gpu: 1
—
# gpu-nodepool-ondemand.yaml – 按需实例节点池（兜底）
apiVersion: apps/v1
kind: Deployment
metadata:
name: agent-worker-ondemand
labels:
app: agent-worker
pool: ondemand
spec:
replicas: 1
template:
spec:
nodeSelector:
node-pool: gpu-ondemand
priorityClassName: high-priority # 高优先级，不会被抢占
containers:
– name: agent
image: agent-worker:latest
resources:
requests:
nvidia.com/gpu: 1
四、边界分析与架构权衡
4.1 冷启动问题
Agent Pod 启动后需要加载模型到 GPU，冷启动时间 10-30 秒。HPA 扩容时新 Pod 无法立即承接流量，导致请求排队。
解决方案：保持最小副本数（minReplicas ≥ 2），避免从零扩容；使用 KEDA 的预热机制，在队列深度接近阈值时提前扩容；模型文件使用 PVC 或 Image Cache 预加载，减少拉取时间。
4.2 缩容时的任务丢失
HPA 缩容时可能杀掉正在处理任务的 Pod。即使有 PreStop Hook，如果任务执行时间超过 terminationGracePeriodSeconds，任务会被强制终止。
解决方案：设置足够长的 terminationGracePeriodSeconds（建议 120s）；使用 Redis Stream 的消费者组机制，未 ACK 的消息会自动重新分配给其他消费者；在 Pod 退出前将未完成任务的状态写入 Redis，新 Pod 启动后恢复。
4.3 GPU 资源碎片化
不同 Agent 需要不同规格的 GPU，长时间运行后节点上可能出现 GPU 碎片——每个节点都有空闲 GPU 但无法满足新 Pod 的需求。
解决方案：统一 GPU 规格（全部使用 T4 或全部使用 A100），减少碎片；使用 Descheduler 定期重平衡 Pod；对 GPU 需求做分级，小模型用 CPU 推理，大模型用 GPU。
4.4 Spot 实例中断
Spot 实例可能被云厂商随时回收，导致 Agent Pod 被驱逐。
解决方案：使用 Spot 实例中断通知（通常提前 2 分钟），在通知触发后停止从队列拉取新任务；按需实例池始终保持最低副本数，确保服务可用性；Agent 设计为无状态，任务状态存储在 Redis 而非本地。
五、总结
K8s 上 Agent 弹性部署的核心不是 HPA 配置本身，而是三个前置条件：任务队列解耦、自定义指标驱动、优雅上下线机制。
没有任务队列，HPA 无法获取真实负载指标；没有自定义指标，基于 CPU 的扩缩容对 Agent 毫无意义；没有优雅退出，缩容时任务丢失会让用户直接看到错误。
从云原生实践的角度，Agent 弹性部署还有一个容易被忽略的维度——成本。GPU 是最贵的计算资源，弹性策略的目标不只是"扛住流量"，更是"用最少的 GPU 跑最多的任务"。Spot 实例 + 按需兜底的分层策略，配合基于队列深度的精确扩缩容，可以让 GPU 成本降低 40-60%。
最后一点：弹性策略不是配置完就不管了。队列深度阈值、扩缩容速率、最小副本数——这些参数需要根据实际业务流量持续调优。建议第一周每天看一次 HPA 事件，确认扩缩容行为符合预期，再逐步放长监控间隔。
