5487 字
约 18 分钟
1
第23章 Agent部署与运维
2026-09-18
第23章 Agent部署与运维
来源：https://ai-agent-guide.xiaofuge.cn/chapters/ch19-deployment.html 所属：第七篇-工程化
从 Demo 到生产——部署、监控、灰度、CI/CD 全流程
23.1 Demo 能跑 ≠ 生产能用
在 Jupyter Notebook 里跑通一个 Agent Demo 只需要 30 分钟。但把它放到生产环境，面对真实用户、真实流量、真实故障，需要解决完全不同的问题。
这个鸿沟比想象中大得多。Demo 阶段的 Agent 跑在开发者的笔记本上：一个用户、一次请求、出错就重来。生产环境的 Agent 则要同时面对几百个并发会话，每个会话可能持续几十轮工具调用，任何一轮 LLM API 抖动、超时或限流都会传导到用户体验。更棘手的是，Agent 的"故障"往往是软性的——服务没挂，但回答质量下降了、步数失控了、Token 费用暴涨了。传统 Web 服务的"进程存活 = 健康"判断在 Agent 上完全不成立。
从 Demo 到生产，本质上是补上六块短板：并发能力（从单进程到多实例负载均衡）、故障韧性（从崩溃重启到重试/降级/熔断）、可观测性（从 print 调试到全链路 Trace）、发布安全（从停机更新到灰度发布）、成本控制（从不管不顾到预算告警）、安全合规（从 Key 硬编码到密钥托管）。下面这张对照表概括了两者的差距： 🧪 Demo 级 Agent
- 单进程、单用户、无并发
- API Key 硬编码在代码里
- 没有日志，出了问题靠 print
- LLM 报错就直接崩溃
- 无限流、无重试、无超时
- 更新代码需要停机
🏭 生产级 Agent
- 多实例、负载均衡、高可用
- 密钥管理（Vault / KMS）
- 全链路日志 + 可观测性
- 优雅降级 + 自动重试 + 超时控制
- 限流 + 熔断 + 步数控制
- 蓝绿部署 / 灰度发布 / 滚动更新
23.2 生产级 Agent 部署架构
生产级 Agent 不是"一个会调 LLM 的 Web 服务"那么简单。它需要把不稳定的 LLM 调用和稳定的业务服务解耦，用一层代理（Proxy）来吸收 LLM 侧的抖动。这样无论底层模型是 OpenAI、Anthropic 还是国产模型，业务层都不需要感知。
下面这套五层架构经过大量生产验证，核心设计原则是：无状态、可替换、可观测。无状态意味着 Agent Service 不保存会话，会话上下文放在 Redis，这样任何实例都能处理任何请求，配合负载均衡实现水平扩展。可替换意味着 LLM Proxy 屏蔽了具体模型，换模型只需改 Proxy 配置。可观测意味着每一层都埋点，Trace ID 贯穿全链路。 🏗️ 推荐架构：API Gateway + Agent Service + LLM Proxy
🌐
API Gateway (Nginx / Kong) SSL 终止、限流、认证、请求路由 🤖
Agent Service (多实例) Python/Java 服务，处理 Agent 逻辑，无状态可水平扩展 🔗
LLM Proxy (自建网关) 多模型路由、Key 轮换、缓存、降级、成本统计
💾
数据层 Redis（会话/缓存）+ PostgreSQL（持久化）+ 向量数据库（RAG） 📊
监控层 Prometheus（指标）+ Grafana（大屏）+ LangSmith/Langfuse（Trace）
Dockerfile 模板
Dockerfile
FROM python:3.11-slim
# 时区
ENV TZ=Asia/Shanghai
RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
# 安装依赖
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制代码
COPY . .
# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
CMD curl -f http://localhost:8080/health || exit 1
# 启动
EXPOSE 8080
CMD ["gunicorn", "--bind", "0.0.0.0:8080", "--workers", "4", \
"--timeout", "120", "--access-logfile", "-", "app:app"]
Docker Compose 部署
docker-compose.yml
version: '3.8'
services:
agent-app:
build: .
container_name: agent-app
restart: on-failure:3 # 最多重启 3 次
ports:
- "8080:8080"
environment:
- REDIS_URL=redis://redis:6379
- DB_URL=postgresql://user:pass@postgres:5432/agent
- LLM_API_KEY=${LLM_API_KEY} # 从 .env 读取
- ENV=production
depends_on:
redis:
condition: service_healthy
postgres:
condition: service_healthy
deploy:
resources:
limits:
memory: 2G
cpus: '2.0'
logging:
driver: "json-file"
options:
max-size: "10m"
max-file: "3"
redis:
image: redis:7-alpine
healthcheck:
test: ["CMD", "redis-cli", "ping"]
postgres:
image: postgres:16
environment:
POSTGRES_DB: agent
POSTGRES_USER: user
POSTGRES_PASSWORD: pass
healthcheck:
test: ["CMD", "pg_isready"]
23.3 CI/CD：持续集成与持续部署
Agent 的 CI/CD 与传统软件不同，需要额外关注Prompt 版本管理和评估回归。
传统软件的变更主要体现在代码上，代码变了行为才变，所以单元测试 + 集成测试就能守住质量底线。但 Agent 的行为由代码 + Prompt + 模型版本三者共同决定——你可能一行代码没改，只是优化了一句 System Prompt，或者上游模型悄悄升级了小版本，Agent 的回答质量就可能退化。而单元测试断言的是确定性输出，对 Agent 的非确定性文本输出几乎无能为力。
因此 Agent 的 CI/CD 流水线必须在"单元测试"和"部署"之间插入一个关键环节：评估回归（Evaluation-Driven Deployment, EDD）。做法是维护一个覆盖核心场景的评估数据集（几十到几百条带标准答案的用例），每次提交都跑一遍，用 LLM-as-Judge 或规则打分。分数低于阈值（如 0.85）就阻断发布。这把"Prompt 改动"提升到了和"代码改动"同等的严肃级别。同时，Prompt 应该像代码一样进版本库、做 Diff 审查，而不是散落在工程师的本地文件里。 🔑 Agent CI/CD 的特殊环节：评估回归
传统 CI 只跑单元测试（assert x == y）。Agent CI 需要额外跑评估数据集（EDD），确保 Prompt 修改没有导致回答质量下降。如果评估分数低于阈值，CI 自动阻止发布。
.github/workflows/agent-ci.yml
name: Agent CI/CD
on:
push:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
# 1. 单元测试
- name: Run Unit Tests
run: | pip install -r requirements.txt
pytest tests/ --cov=app --cov-report=xml
# 2. Agent 评估回归（关键！）
- name: Run Evaluation Regression
env:
LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
run: | python -m eval.run_eval \
--dataset eval/datasets/prod.json \
--threshold 0.85 \
--report eval/report.json
# 如果评估分数 80% | | **应用层** | QPS/延迟/错误率 | Prometheus + Grafana | P99 > 10s | | **Agent层** | 成功率/步数/Token消耗 | LangSmith / Langfuse | 成功率 预算 80% | | **业务层** | 用户满意度/投诉率 | 反馈系统 | 差评率 > 5% | ```
from prometheus_client import Counter, Histogram, Gauge
# Agent 指标定义
agent_requests = Counter(
'agent_requests_total',
'Total agent requests',
['chapter', 'status'] # 标签：章节、状态(success/fail)
)
agent_latency = Histogram(
'agent_latency_seconds',
'Agent response latency',
['chapter'],
buckets=[0.5, 1, 2, 5, 10, 30, 60, 120] # 延迟分桶
)
agent_tokens = Counter(
'agent_tokens_total',
'Total tokens consumed',
['model', 'type'] # 模型、类型(input/output)
)
agent_steps = Histogram(
'agent_steps_count',
'Number of steps per request',
buckets=[1, 3, 5, 10, 15, 20]
)
# 在 Agent 执行流程中埋点
def handle_request(user_input, chapter):
start = time.time()
try:
result = agent.invoke(user_input)
agent_requests.labels(chapter=chapter, status='success').inc()
return result
except Exception as e:
agent_requests.labels(chapter=chapter, status='fail').inc()
raise
finally:
agent_latency.labels(chapter=chapter).observe(time.time() - start)
import { Counter, Histogram, register } from 'prom-client';
// Agent 指标定义
const agentRequests = new Counter({
name: 'agent_requests_total',
help: 'Total agent requests',
labelNames: ['chapter', 'status'],
});
const agentLatency = new Histogram({
name: 'agent_latency_seconds',
help: 'Agent response latency',
labelNames: ['chapter'],
buckets: [0.5, 1, 2, 5, 10, 30, 60, 120],
});
const agentTokens = new Counter({
name: 'agent_tokens_total',
help: 'Total tokens consumed',
labelNames: ['model', 'type'],
});
const agentSteps = new Histogram({
name: 'agent_steps_count',
help: 'Number of steps per request',
buckets: [1, 3, 5, 10, 15, 20],
});
// 在 Agent 执行流程中埋点
async function handleRequest(userInput: string, chapter: string) {
const start = Date.now();
try {
const result = await agent.invoke(userInput);
agentRequests.labels({ chapter, status: 'success' }).inc();
return result;
} catch (e) {
agentRequests.labels({ chapter, status: 'fail' }).inc();
throw e;
} finally {
agentLatency.labels({ chapter }).observe((Date.now() - start) / 1000);
}
}
package main
import (
"time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
// Agent 指标定义
var (
agentRequests = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "agent_requests_total",
Help: "Total agent requests",
}, []string{"chapter", "status"})
agentLatency = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "agent_latency_seconds",
Help: "Agent response latency",
Buckets: []float64{0.5, 1, 2, 5, 10, 30, 60, 120},
}, []string{"chapter"})
agentTokens = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "agent_tokens_total",
Help: "Total tokens consumed",
}, []string{"model", "type"})
agentSteps = promauto.NewHistogram(prometheus.HistogramOpts{
Name: "agent_steps_count",
Help: "Number of steps per request",
Buckets: []float64{1, 3, 5, 10, 15, 20},
})
)
// 在 Agent 执行流程中埋点
func handleRequest(userInput, chapter string) (interface{}, error) {
start := time.Now()
result, err := agent.Invoke(userInput)
if err != nil {
agentRequests.WithLabelValues(chapter, "fail").Inc()
return nil, err
}
agentRequests.WithLabelValues(chapter, "success").Inc()
agentLatency.WithLabelValues(chapter).Observe(time.Since(start).Seconds())
return result, nil
}
import io.prometheus.client.*;
import java.time.Instant;
import java.time.Duration;
public class AgentMetrics {
// Agent 指标定义
static final Counter agentRequests = Counter.build()
.name("agent_requests_total")
.help("Total agent requests")
.labelNames("chapter", "status")
.register();
static final Histogram agentLatency = Histogram.build()
.name("agent_latency_seconds")
.help("Agent response latency")
.labelNames("chapter")
.buckets(0.5, 1, 2, 5, 10, 30, 60, 120)
.register();
static final Counter agentTokens = Counter.build()
.name("agent_tokens_total")
.help("Total tokens consumed")
.labelNames("model", "type")
.register();
static final Histogram agentSteps = Histogram.build()
.name("agent_steps_count")
.help("Number of steps per request")
.buckets(1, 3, 5, 10, 15, 20)
.register();
// 在 Agent 执行流程中埋点
public Object handleRequest(String userInput, String chapter) {
long start = System.currentTimeMillis();
try {
Object result = agent.invoke(userInput);
agentRequests.labels(chapter, "success").inc();
return result;
} catch (Exception e) {
agentRequests.labels(chapter, "fail").inc();
throw e;
} finally {
double elapsed = (System.currentTimeMillis() - start) / 1000.0;
agentLatency.labels(chapter).observe(elapsed);
}
}
}
23.6 成本优化策略
Agent 的最大成本是 LLM API 调用费用。一个不加优化的 Agent，月账单可能从 $100 飙到 $10,000。
💰 五大成本优化策略
1. 模型分级路由 简单问题用 mini 模型（$0.15/M），复杂问题才用旗舰模型（$5/M）。80% 问题用 mini 解决，成本降 5-10 倍。
2. 语义缓存 用向量相似度判断"是否问过类似问题"，命中缓存直接返回，不再调用 LLM。命中率 20-40%。
3. Prompt 压缩 精简 System Prompt，去掉不必要的示例和说明。每省 1000 Token = 每次省 $0.01-0.05。
4. 步数控制 限制 Agent 最大步数（如 10 步），避免无限循环烧 Token。每多一步 = 一次完整 LLM 调用。
5. 批量处理 多个独立任务合并为一次 LLM 调用（batch）。OpenAI Batch API 半价，适合非实时场景。
def route_model(query: str, has_tools: bool) -> str:
"""根据问题复杂度选择模型"""
# 简单问答 → mini 模型（便宜 30 倍）
if not has_tools and len(query) str | None:
query_vec = embeddings.embed_query(query)
for cached_vec, response in self.cache.items():
similarity = cosine_similarity(query_vec, cached_vec)
if similarity > self.threshold:
return response # 缓存命中
return None # 未命中
def set(self, query: str, response: str):
vec = embeddings.embed_query(query)
self.cache[vec] = response
# 使用
cache = SemanticCache(threshold=0.95)
def chat(query):
# 先查缓存
cached = cache.get(query)
if cached:
return cached # 命中，不调用 LLM
# 未命中，调用 LLM
response = llm.invoke(query)
cache.set(query, response)
return response
import { OpenAIEmbeddings } from '@langchain/openai';
/** 基于向量相似度的缓存 */
class SemanticCache {
private cache: Map = new Map();
private threshold: number;
private embeddings: OpenAIEmbeddings;
constructor(threshold = 0.95, embeddings: OpenAIEmbeddings) {
this.threshold = threshold;
this.embeddings = embeddings;
}
async get(query: string): Promise {
const queryVec = await this.embeddings.embedQuery(query);
for (const [cachedVec, response] of this.cache.entries()) {
const similarity = cosineSimilarity(queryVec, cachedVec);
if (similarity > this.threshold) {
return response; // 缓存命中
}
}
return null; // 未命中
}
async set(query: string, response: string): Promise {
const vec = await this.embeddings.embedQuery(query);
this.cache.set(vec, response);
}
}
function cosineSimilarity(a: number[], b: number[]): number {
let dot = 0, normA = 0, normB = 0;
for (let i = 0; i {
// 先查缓存
const cached = await cache.get(query);
if (cached) {
return cached; // 命中，不调用 LLM
}
// 未命中，调用 LLM
const response = await llm.invoke(query);
await cache.set(query, response);
return response;
}
package main
import (
"math"
)
// SemanticCache 基于向量相似度的缓存
type SemanticCache struct {
cache map[[128]float32]string
threshold float64
embeddings EmbeddingsClient
}
func NewSemanticCache(threshold float64, emb EmbeddingsClient) *SemanticCache {
return &SemanticCache{
cache: make(map[[128]float32]string),
threshold: threshold,
embeddings: emb,
}
}
// Get 查询缓存
func (c *SemanticCache) Get(query string) (string, bool) {
queryVec := c.embeddings.EmbedQuery(query)
for cachedVec, response := range c.cache {
sim := cosineSimilarity(queryVec, cachedVec[:])
if sim > c.threshold {
return response, true // 缓存命中
}
}
return "", false // 未命中
}
// Set 设置缓存
func (c *SemanticCache) Set(query, response string) {
vec := c.embeddings.EmbedQuery(query)
var arr [128]float32
copy(arr[:], vec)
c.cache[arr] = response
}
func cosineSimilarity(a, b []float32) float64 {
var dot, normA, normB float64
for i := range a {
dot += float64(a[i] * b[i])
normA += float64(a[i] * a[i])
normB += float64(b[i] * b[i])
}
return dot / (math.Sqrt(normA) * math.Sqrt(normB))
}
import java.util.*;
public class SemanticCache {
/** 基于向量相似度的缓存 */
private Map- , String> cache = new HashMap<>();
private double threshold;
private EmbeddingsClient embeddings;
public SemanticCache(double threshold, EmbeddingsClient embeddings) {
this.threshold = threshold;
this.embeddings = embeddings;
}
/** 查询缓存 */
public Optional get(String query) {
List queryVec = embeddings.embedQuery(query);
for (Map.Entry, String> entry : cache.entrySet()) {
double similarity = cosineSimilarity(queryVec, entry.getKey());
if (similarity > threshold) {
return Optional.of(entry.getValue()); // 缓存命中
}
}
return Optional.empty(); // 未命中
}
/** 设置缓存 */
public void set(String query, String response) {
List vec = embeddings.embedQuery(query);
cache.put(vec, response);
}
private double cosineSimilarity(List a, List b) {
double dot = 0, normA = 0, normB = 0;
for (int i = 0; i
**📋 八股总结 — 面试高频考点**
Q1: Demo 级 Agent 和生产级 Agent 的区别是什么？
Demo 关注功能跑通：单进程、无并发、无监控、无重试。生产级关注可靠运行：多实例高可用、限流熔断、全链路监控、优雅降级、灰度发布、成本控制。**面试关键**：能说出 Demo 的 5 个缺陷和对应的生产级解决方案（密钥管理、日志监控、错误重试、超时控制、资源限制）。
Q2: Agent 的 CI/CD 和传统软件有什么不同？
最大的区别是**评估回归**。传统 CI 跑单元测试（assert），Agent CI 额外要跑评估数据集（EDD），确保 Prompt 修改没有导致回答质量下降。如果评估分数低于阈值（如 0.85），CI 自动阻止发布。这是因为 Agent 输出有随机性，代码逻辑没变但 Prompt 变了可能导致行为退化。
Q3: 什么是金丝雀发布？Agent 为什么需要它？
金丝雀发布：新版本先承接 5-10% 流量，观察指标正常后再全量。Agent 需要它因为：① LLM 输出有随机性，测试集无法覆盖所有情况；② Prompt 微调可能有副作用；③ 新工具可能有未知的失败模式。金丝雀发布让问题影响范围最小化，发现问题可以快速回滚。
Q4: Agent 的监控体系应该包含哪些层面？
五层监控：① **系统层**（CPU/内存/磁盘）；② **应用层**（QPS/延迟/错误率）；③ **Agent层**（成功率/步数/Token消耗，用 LangSmith/Langfuse）；④ **LLM层**（API延迟/配额/费用）；⑤ **业务层**（用户满意度/投诉率）。每层设定告警阈值，实现 5 分钟发现-定位-恢复。
Q5: 如何优化 Agent 的 LLM API 成本？
① **模型分级路由**：80% 简单问题用 mini 模型（$0.15/M），复杂问题才用旗舰模型，成本降 5-10 倍；② **语义缓存**：相似问题命中缓存直接返回，命中率 20-40%；③ **Prompt 压缩**：精简 System Prompt；④ **步数控制**：限制最大步数防止烧 Token；⑤ **批量处理**：非实时任务用 Batch API（半价）。
## 23.7 Kubernetes 部署：企业级容器编排
当 Agent 服务从单机 Docker 走向多实例、多环境时，Kubernetes（K8s）成为企业级部署的标配。K8s 提供了**自动扩缩容、自愈、滚动更新、服务发现**等生产级能力，是 Agent 从"能跑"到"能扛流量"的关键基础设施。
### 23.7.1 Agent 的 K8s 部署模板
YAML
Python (kubectl)
```yaml
# agent-deployment.yaml — Agent 服务部署
apiVersion: apps/v1
kind: Deployment
metadata:
name: ai-agent-service
namespace: production
labels:
app: ai-agent
tier: backend
spec:
replicas: 3 # 3 副本保证高可用
selector:
matchLabels:
app: ai-agent
strategy:
type: RollingUpdate # 滚动更新
rollingUpdate:
maxSurge: 1 # 滚动更新时最多多出 1 个 Pod
maxUnavailable: 0 # 滚动更新时不允许减少可用 Pod
template:
metadata:
labels:
app: ai-agent
spec:
containers:
- name: agent
image: registry.cn-beijing.aliyuncs.com/myorg/ai-agent:v1.2.0
ports:
- containerPort: 8080
env:
- name: LLM_API_KEY
valueFrom:
secretKeyRef:
name: llm-secrets
key: api-key
- name: REDIS_URL
value: "redis://redis-cluster:6379"
- name: MAX_CONCURRENT_AGENTS
value: "50"
resources:
requests: # 资源请求（调度依据）
cpu: "500m"
memory: "1Gi"
limits: # 资源上限（防失控）
cpu: "2000m"
memory: "4Gi"
livenessProbe: # 存活探针：挂了自动重启
httpGet:
path: /health
port: 8080
initialDelaySeconds: 30
periodSeconds: 10
readinessProbe: # 就绪探针：没准备好不接流量
httpGet:
path: /ready
port: 8080
initialDelaySeconds: 10
periodSeconds: 5
---
# HPA — 自动扩缩容
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: ai-agent-hpa
namespace: production
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: ai-agent-service
minReplicas: 3
maxReplicas: 20 # 最多扩到 20 个 Pod
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70 # CPU 超 70% 自动扩容
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: 80
# Python 部署脚本 — 通过 kubectl 管理 Agent 集群
import subprocess
import yaml
def deploy_agent_cluster():
"""一键部署 Agent 集群到 K8s"""
# 1. 应用部署配置
subprocess.run([
"kubectl", "apply", "-f", "agent-deployment.yaml"
], check=True)
# 2. 等待 Pod 就绪
subprocess.run([
"kubectl", "rollout", "status",
"deployment/ai-agent-service",
"-n", "production",
"--timeout=300s"
], check=True)
# 3. 验证服务可用
result = subprocess.run([
"kubectl", "get", "pods", "-n", "production",
"-l", "app=ai-agent",
"-o", "jsonpath={.items[*].status.phase}"
], capture_output=True, text=True)
phases = result.stdout.split()
ready_count = sum(1 for p in phases if p == "Running")
print(f"✅ {ready_count}/{len(phases)} Pods Running")
if ready_count
### 23.7.2 Agent 专属 K8s 配置要点
#### ❌ 普通Web服务的K8s配置
▸ 固定 2 副本，无需弹性扩缩
▸ 请求/响应毫秒级，无长连接
▸ CPU 密集型，内存固定
▸ 健康检查：HTTP 200 即可
▸ 无状态，Pod 可随意重启
#### ✅ Agent 服务的K8s配置
▸ HPA 弹性扩缩（Agent 调用耗时不确定）
▸ SSE 长连接，需要 sticky session
▸ 内存波动大（上下文 + 记忆检索）
▸ 就绪探针需检查 LLM API 连通性
▸ 有状态（会话上下文），需 graceful shutdown
**Agent 优雅关闭（Graceful Shutdown）**：K8s 删除 Pod 时默认给 30 秒宽限期。但 Agent 可能正在执行一个 5 分钟的工具调用。解决方案：① 设置 `terminationGracePeriodSeconds: 300`（5分钟）；② 在 `preStop` 钩子中通知 Agent "正在关闭，请完成当前任务"；③ 配合就绪探针摘除流量，新请求不再进来。
### 23.7.3 Prometheus + Grafana 监控体系
Prometheus 负责指标采集，Grafana 负责可视化，二者组合是云原生监控的事实标准。下面这张架构图展示了 Agent 服务的完整监控数据流：
**📊 Agent 监控体系架构**
Agent 服务
多实例(K8s)
/metrics 端点
Prometheus
定时拉取指标
Grafana
可视化大屏
Langfuse/LangSmith
Agent Trace
Alertmanager
阈值告警
通知渠道
飞书/钉钉/邮件
指标采集 → 可视化 → 阈值告警 → 通知，Trace 走独立的 Agent 观测链路
Agent 服务需要暴露以下自定义指标：
Python
TypeScript
```python
# agent_metrics.py — Agent 自定义 Prometheus 指标
from prometheus_client import Counter, Histogram, Gauge, generate_latest
# === Agent 核心指标 ===
# 1. Agent 执行总次数（按状态分）
agent_requests = Counter(
'agent_requests_total',
'Total agent requests',
['status', 'agent_type'] # status: success/fail/timeout
)
# 2. Agent 执行耗时分布
agent_duration = Histogram(
'agent_duration_seconds',
'Agent execution duration',
['agent_type'],
buckets=[0.5, 1, 2, 5, 10, 30, 60, 120, 300] # 秒
)
# 3. 当前活跃 Agent 数量
active_agents = Gauge(
'active_agents',
'Currently running agents'
)
# 4. LLM API 调用次数与耗时
llm_api_calls = Counter(
'llm_api_calls_total',
'LLM API calls',
['provider', 'model', 'status']
)
# 5. Token 消耗总量
token_consumption = Counter(
'agent_tokens_total',
'Token consumption',
['type', 'model'] # type: input/output
)
# 6. 工具调用次数（按工具名分）
tool_calls = Counter(
'agent_tool_calls_total',
'Tool calls by agent',
['tool_name', 'status']
)
# === 使用示例 ===
import time
def run_agent(task):
active_agents.inc() # 活跃数 +1
start = time.time()
try:
result = execute_agent(task)
agent_requests.labels(status='success', agent_type=task.type).inc()
agent_duration.labels(agent_type=task.type).observe(time.time() - start)
return result
except TimeoutError:
agent_requests.labels(status='timeout', agent_type=task.type).inc()
raise
except Exception as e:
agent_requests.labels(status='fail', agent_type=task.type).inc()
raise
finally:
active_agents.dec() # 活跃数 -1
// agent-metrics.ts — Prometheus client for Node.js
import { Counter, Histogram, Gauge, register } from 'prom-client';
// Agent 核心指标定义
const agentRequests = new Counter({
name: 'agent_requests_total',
help: 'Total agent requests',
labelNames: ['status', 'agent_type'] as const,
});
const agentDuration = new Histogram({
name: 'agent_duration_seconds',
help: 'Agent execution duration',
labelNames: ['agent_type'] as const,
buckets: [0.5, 1, 2, 5, 10, 30, 60, 120, 300],
});
const activeAgents = new Gauge({
name: 'active_agents',
help: 'Currently running agents',
});
const tokenConsumption = new Counter({
name: 'agent_tokens_total',
help: 'Token consumption',
labelNames: ['type', 'model'] as const,
});
// 中间件：自动记录 Agent 执行指标
export function withMetrics Promise>(
fn: T,
agentType: string
): T {
return (async (...args: Parameters) => {
activeAgents.inc();
const start = Date.now();
try {
const result = await fn(...args);
agentRequests.inc({ status: 'success', agent_type: agentType });
return result;
} catch (err) {
agentRequests.inc({
status: err instanceof TimeoutError ? 'timeout' : 'fail',
agent_type: agentType
});
throw err;
} finally {
agentDuration.observe({ agent_type: agentType }, (Date.now() - start) / 1000);
activeAgents.dec();
}
}) as T;
}
对应的 Grafana 告警规则：
# alerting-rules.yml — Prometheus 告警规则
groups:
- name: agent-alerts
rules:
# Agent 成功率低于 95%
- alert: AgentHighErrorRate
expr: | sum(rate(agent_requests_total{status="fail"}[5m])) /
sum(rate(agent_requests_total[5m])) > 0.05
for: 5m
labels:
severity: critical
annotations:
summary: "Agent 错误率超过 5%"
description: "最近 5 分钟 Agent 失败率: {{ $value | humanizePercentage }}"
# Agent P95 响应时间超过 30 秒
- alert: AgentSlowResponse
expr: | histogram_quantile(0.95,
sum(rate(agent_duration_seconds_bucket[5m])) by (le)
) > 30
for: 10m
labels:
severity: warning
annotations:
summary: "Agent P95 响应时间超过 30 秒"
# Token 消耗速率异常（可能是死循环）
- alert: TokenConsumptionSpike
expr: | rate(agent_tokens_total[5m]) > 100000
for: 5m
labels:
severity: critical
annotations:
summary: "Token 消耗速率异常，可能存在 Agent 死循环"
23.7.4 SLA 保障体系
企业级 Agent 服务需要明确的 SLA（Service Level Agreement）承诺。Agent 的 SLA 与传统 Web 服务不同，需要同时保证可用性和质量两个维度。 | SLA 指标 | 目标值 | 测量方式 | 保障手段 | | --- | --- | --- | --- | | 服务可用性 | 99.9%（月级） | HTTP 200 响应率 | 多副本 + 自动故障转移 | | 响应延迟 P95 | 92% | Agent 完成任务且用户满意 | 评估驱动开发 + 回归测试 | | Token 成本/请求 | SLA 计算公式： 月度可用性 = (总分钟数 - 不可用分钟数) / 总分钟数 × 100% 99.9% = 每月最多停机 43.2 分钟。Agent 服务因 LLM API 依赖第三方，建议在 SLA 中明确"LLM API 不可用不计入停机时间"。
23.8 企业级部署 Checklist
以下是从 Demo 到生产环境的完整检查清单，每一条都必须打勾才能上线：
🏗️ 基础设施
☐ K8s 部署模板（Deployment + HPA + Service）
- ☐ 多副本部署（至少 3 副本，跨可用区分布）
- ☐ 配置中心（Secret 管理 API Key，不硬编码）
- ☐ 私有镜像仓库（不依赖 Docker Hub）
📊 监控告警
- ☐ Prometheus 指标采集（自定义 Agent 指标）
- ☐ Grafana 可视化面板（QPS/延迟/成功率/Token消耗）
- ☐ 告警规则（错误率/P95延迟/Token异常）
- ☐ 日志聚合（ELK/Loki，支持 Trace ID 追踪）
🔒 安全合规
- ☐ API Key 通过 Secret 注入，不进镜像
- ☐ HTTPS 全链路加密
- ☐ 审计日志（谁在什么时候调用了什么 Agent）
- ☐ 数据脱敏（用户 PII 不入 LLM）
🔄 发布运维
- ☐ CI/CD 流水线（代码提交→测试→构建→部署）
- ☐ 灰度发布（金丝雀 10%→50%→100%）
- ☐ 回滚机制（一键回滚到上一版本）
- ☐ Runbook（常见故障的处理手册）
💰 成本控制
- ☐ 模型分级路由（简单任务用 mini 模型）
- ☐ 语义缓存（命中率监控）
- ☐ Token 预算告警（日/月级别）
- ☐ 步数限制（防死循环烧钱）
评论
0 条
还没有评论，先写一条吧。