AI Agent 的部署与运维：从开发环境到生产环境的完整指南

本文承接评测与质量保障，系统讲解 AI Agent 的部署与运维体系。文章介绍从开发环境到生产环境的部署架构，讲解 LLM 推理服务的部署与优化、Agent 运行时的状态管理与任务调度、弹性扩缩容与成本优化、可观测性三大支柱（日志/指标/链路追踪）、故障处理与高可用设计、多环境管理与发布策略，并结合配置示例和代码说明如何把一个 Agent 系统稳定、高效、低成本地运行在生产环境中。

目录

• 一、从开发环境到生产环境：部署架构总览
• 二、模型服务化：LLM 推理服务的部署与优化
• 三、Agent 运行时架构：状态管理、任务调度与执行引擎
• 四、弹性扩缩容与成本优化
• 五、可观测性：日志、指标与链路追踪
• 六、故障处理与高可用设计
• 七、多环境管理与发布策略
• 八、上线前检查清单
• 结语

引言

当你在笔记本电脑上把一个 AI Agent 跑通时，距离它在生产环境稳定运行还有很长的路要走。开发环境里，你一个人用、数据量小、没有并发、出了问题直接重启就行。但生产环境完全不同：成百上千的用户同时使用、任务 7×24 小时不间断运行、任何一次宕机都可能造成业务损失、成本账单每月都在增长。

部署与运维是 Agent 工程化中最容易被低估的环节。很多团队把大量精力花在提示词调优和功能开发上，却在上线后被各种运维问题困扰：响应时间忽快忽慢、高峰期服务崩溃、Token 成本失控、出了问题找不到根因、版本更新导致服务中断。

本文从部署架构、模型服务、运行时、扩缩容、可观测性、高可用、多环境管理七个层面，系统讲解如何把一个 AI Agent 从开发环境部署到生产环境，并持续稳定高效地运行。

一、从开发环境到生产环境：部署架构总览

1.1 开发环境与生产环境的本质差异

很多人以为部署就是”把代码放到服务器上运行”，但实际上开发环境和生产环境的差异是系统性的：

这些差异决定了生产环境的架构必须在可用性、性能、成本、安全四个维度上做专门的设计和优化。

1.2 典型生产部署架构

一个典型的 AI Agent 生产部署架构包含以下层次：

用户层
  ├── Web 前端 / 移动端 / API 接入
  └── 负载均衡（Nginx / ALB）
        │
网关层
  ├── API 网关（鉴权、限流、路由）
  └── 消息队列（异步任务、削峰填谷）
        │
应用层
  ├── Agent 服务（无状态，水平扩展）
  │     ├── 任务调度器
  │     ├── 执行引擎
  │     └── 工具调用层
  ├── 工具服务（搜索、数据库、第三方 API）
  └── 业务服务（用户、订单、权限等）
        │
模型层
  ├── LLM 推理服务（vLLM / TGI / 云端 API）
  ├── Embedding 服务
  └── 模型缓存（Redis / 本地缓存）
        │
数据层
  ├── 向量数据库（Milvus / Pinecone）
  ├── 关系型数据库（PostgreSQL / MySQL）
  ├── 缓存（Redis）
  └── 对象存储（S3 / OSS）
        │
运维层
  ├── 监控告警（Prometheus + Grafana）
  ├── 日志收集（ELK / Loki）
  ├── 链路追踪（Jaeger / SkyWalking）
  └── CI/CD（GitHub Actions / Jenkins）

这个架构的核心设计原则是：无状态应用层水平扩展，有状态数据层独立托管，模型层单独优化，运维层全面覆盖。

1.3 容器化与编排

生产环境的 Agent 服务应该容器化部署，使用 Kubernetes 或类似的容器编排平台管理。

容器化的好处： – 环境一致性：开发、测试、生产环境使用相同的容器镜像，避免”在我机器上能跑”的问题 – 快速部署：新版本发布只需拉取新镜像、滚动更新，分钟级完成 – 弹性伸缩：根据负载自动增减容器实例 – 故障自愈：容器崩溃后自动重启，节点故障后自动迁移 – 资源隔离：不同服务的资源使用互相隔离，避免争抢

Kubernetes 关键配置：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: agent-service
  template:
    metadata:
      labels:
        app: agent-service
    spec:
      containers:
      - name: agent
        image: registry.example.com/agent-service:v1.2.0
        ports:
        - containerPort: 8000
        resources:
          requests:
            cpu: "500m"
            memory: "1Gi"
          limits:
            cpu: "2000m"
            memory: "4Gi"
        env:
        - name: MODEL_API_KEY
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: model-api-key
        readinessProbe:
          httpGet:
            path: /health/ready
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
        livenessProbe:
          httpGet:
            path: /health/live
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10

关键配置说明： – replicas: 3：至少 3 个实例，确保高可用 – resources.requests/limits：设置资源请求和上限，避免资源争抢和 OOM – readinessProbe：就绪探针，只有通过健康检查的实例才接收流量 – livenessProbe：存活探针，实例异常时自动重启 – secretKeyRef：敏感信息从 Secret 中读取，不硬编码在配置里

二、模型服务化：LLM 推理服务的部署与优化

2.1 模型服务的三种部署方式

LLM 推理服务是 Agent 系统中最核心、最昂贵的组件。有三种部署方式：

对于大多数团队，建议从云端 API 开始，验证业务价值后再考虑自建模型。如果有大量简单任务（如分类、摘要），可以用小模型自建处理，复杂任务走云端 API，降低整体成本。

2.2 自建模型推理服务优化

如果选择自建开源模型，推理服务的性能优化至关重要。常用的推理框架包括 vLLM、TGI（Text Generation Inference）、TensorRT-LLM 等。

vLLM 部署示例：

# 启动 vLLM 推理服务
python -m vllm.entrypoints.openai.api_server \
  --model /models/Qwen2-72B-Instruct \
  --tensor-parallel-size 4 \
  --gpu-memory-utilization 0.9 \
  --max-model-len 32768 \
  --enable-prefix-caching \
  --port 8000

关键优化参数： – --tensor-parallel-size 4：张量并行，用 4 张 GPU 同时推理一个大模型 – --gpu-memory-utilization 0.9：GPU 显存利用率，0.9 表示用 90% 显存做 KV Cache – --max-model-len 32768：最大上下文长度，根据业务需求设置 – --enable-prefix-caching：启用前缀缓存，相同前缀的请求复用 KV Cache，大幅提升吞吐量

推理性能优化的核心指标： – 吞吐量（Throughput）：每秒处理的 Token 数，越高越好 – 首 Token 延迟（TTFT）：从请求发出到收到第一个 Token 的时间 – 单 Token 延迟（TPOT）：每个后续 Token 的生成时间 – GPU 利用率：GPU 的计算和显存利用率，避免闲置

2.3 模型缓存策略

LLM 推理是 Agent 系统中最大的成本项。合理使用缓存可以显著降低成本和延迟。

缓存层级：

应用层缓存（Redis）
  ├── 完全相同的请求 → 直接返回缓存结果
  └── 相同前缀的请求 → 复用前缀的 KV Cache

推理层缓存（vLLM Prefix Caching）
  └── 自动管理 KV Cache，相同前缀复用

CDN/边缘缓存
  └── 静态资源、Embedding 结果缓存

应用层缓存实现示例：

import hashlib
import json
import redis

redis_client = redis.Redis(host="localhost", port=6379, db=0)

def get_cache_key(messages: list, model: str, temperature: float) -> str:
    """生成缓存键"""
    content = json.dumps({
        "messages": messages,
        "model": model,
        "temperature": temperature,
    }, sort_keys=True)
    return f"llm_cache:{hashlib.md5(content.encode()).hexdigest()}"

def cached_llm_call(messages: list, model: str, temperature: float = 0.7):
    """带缓存的 LLM 调用"""
    # 只有 temperature=0 的确定性请求才缓存
    if temperature > 0:
        return direct_llm_call(messages, model, temperature)

    cache_key = get_cache_key(messages, model, temperature)
    cached = redis_client.get(cache_key)

    if cached:
        return json.loads(cached)

    result = direct_llm_call(messages, model, temperature)

    # 缓存 24 小时
    redis_client.setex(cache_key, 86400, json.dumps(result))

    return result

缓存注意事项： – 只缓存 temperature=0 的确定性请求，非确定性请求缓存没有意义 – 设置合理的过期时间，避免模型更新后返回过时结果 – 监控缓存命中率，命中率低说明缓存策略需要优化 – 敏感数据不要缓存，或加密后缓存

2.4 模型路由与降级

在生产环境中，不要把所有请求都发给同一个模型。根据任务复杂度和延迟要求，智能路由到不同模型：

简单任务（分类、提取、格式化）→ 小模型（Qwen-7B / GPT-3.5）
中等任务（摘要、翻译、常规问答）→ 中模型（Qwen-32B / GPT-4-mini）
复杂任务（推理、规划、代码生成）→ 大模型（GPT-4 / Claude-3-Opus）

降级策略：当主模型不可用或超时时，自动降级到备用模型：

async def robust_llm_call(messages, model="gpt-4", max_retries=2):
    """带降级的 LLM 调用"""
    models = ["gpt-4", "gpt-4-mini", "qwen-72b"]

    for i, m in enumerate(models):
        try:
            result = await llm_call(messages, model=m, timeout=30)
            if i > 0:
                log.warning(f"降级到模型 {m}，原模型 {model} 不可用")
            return result
        except (TimeoutError, APIError) as e:
            log.error(f"模型 {m} 调用失败: {e}")
            continue

    raise RuntimeError("所有模型均不可用")

三、Agent 运行时架构：状态管理、任务调度与执行引擎

3.1 Agent 运行时的核心组件

Agent 运行时是执行 Agent 任务的核心引擎，包含以下组件：

• 任务调度器：接收任务、分配资源、管理任务生命周期
• 执行引擎：执行 Agent 的决策循环（思考→行动→观察）
• 状态管理器：管理 Agent 的会话状态、上下文、记忆
• 工具调用层：统一管理工具的注册、调用、超时、重试
• 事件总线：组件间的异步通信和事件通知

3.2 状态管理：有状态 vs 无状态

Agent 服务的状态管理是架构设计的关键决策。

无状态设计：每个请求独立，不保存会话状态。状态由客户端或外部存储维护。 – 优点：水平扩展简单、故障恢复快、资源利用率高 – 缺点：每次请求需要重新加载上下文、长任务需要外部状态存储

有状态设计：服务端维护会话状态，长任务在服务端持续执行。 – 优点：上下文管理方便、长任务执行流畅 – 缺点：扩展复杂、会话粘滞、故障恢复困难

推荐的混合方案：Agent 服务本身无状态，会话状态存储在外部（Redis/数据库），长任务通过消息队列异步执行。

会话状态存储示例：

import json
import redis
from dataclasses import dataclass, asdict
from typing import Optional

redis_client = redis.Redis(host="localhost", port=6379, db=1)

@dataclass
class AgentSession:
    session_id: str
    user_id: str
    messages: list
    memory: dict
    current_task: Optional[dict]
    created_at: float
    updated_at: float

class SessionManager:
    def __init__(self, ttl: int = 86400):
        self.ttl = ttl  # 会话过期时间，默认 24 小时

    def _key(self, session_id: str) -> str:
        return f"agent_session:{session_id}"

    def save(self, session: AgentSession):
        """保存会话状态"""
        import time
        session.updated_at = time.time()
        data = json.dumps(asdict(session))
        redis_client.setex(self._key(session.session_id), self.ttl, data)

    def load(self, session_id: str) -> Optional[AgentSession]:
        """加载会话状态"""
        data = redis_client.get(self._key(session_id))
        if not data:
            return None
        return AgentSession(**json.loads(data))

    def delete(self, session_id: str):
        """删除会话"""
        redis_client.delete(self._key(session_id))

3.3 任务调度与异步执行

用户提交的 Agent 任务可能需要执行几分钟甚至几十分钟。同步等待会占用连接、超时失败。正确的做法是异步执行：

用户提交任务 → API 接收 → 写入消息队列 → 立即返回任务 ID
                                      ↓
                            工作进程从队列取任务 → 执行 Agent → 更新任务状态
                                      ↓
用户轮询 / WebSocket 推送 ← 任务状态存储（Redis/DB）← 执行完成

任务状态机：

PENDING（排队中）→ RUNNING（执行中）→ SUCCESS（成功）
                          ↓
                      FAILED（失败）
                          ↓
                      RETRYING（重试中）→ RUNNING

异步任务执行示例：

from celery import Celery
import time

app = Celery("agent_tasks", broker="redis://localhost:6379/0")

class TaskStatus:
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"

def update_task_status(task_id: str, status: str, result: dict = None, error: str = None):
    """更新任务状态"""
    data = {
        "task_id": task_id,
        "status": status,
        "result": result,
        "error": error,
        "updated_at": time.time(),
    }
    redis_client.setex(f"agent_task:{task_id}", 86400, json.dumps(data))

@app.task(bind=True, max_retries=3)
def execute_agent_task(self, task_id: str, task_input: dict):
    """执行 Agent 任务（Celery 异步任务）"""
    try:
        update_task_status(task_id, TaskStatus.RUNNING)

        # 初始化 Agent
        agent = Agent(
            session_id=task_input["session_id"],
            tools=task_input.get("tools", []),
        )

        # 执行任务
        result = agent.run(task_input["prompt"])

        update_task_status(task_id, TaskStatus.SUCCESS, result=result)
        return result

    except Exception as e:
        if self.request.retries < self.max_retries:
            update_task_status(task_id, "retrying", error=str(e))
            raise self.retry(countdown=10 * (self.request.retries + 1))
        else:
            update_task_status(task_id, TaskStatus.FAILED, error=str(e))
            raise

3.4 工具调用层的统一管理

Agent 可能调用十几个甚至几十个工具。统一的工具调用层可以处理超时、重试、限流、熔断、日志等横切关注点。

工具调用层设计：

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Optional
import time
import logging

logger = logging.getLogger(__name__)

@dataclass
class ToolResult:
    success: bool
    data: Any = None
    error: str = ""
    duration_ms: float = 0

class Tool(ABC):
    def __init__(self, name: str, timeout: int = 30, max_retries: int = 2):
        self.name = name
        self.timeout = timeout
        self.max_retries = max_retries

    @abstractmethod
    def execute(self, params: dict) -> Any:
        """工具的具体执行逻辑"""
        pass

    def call(self, params: dict) -> ToolResult:
        """带超时、重试、日志的统一调用入口"""
        start = time.time()

        for attempt in range(self.max_retries + 1):
            try:
                data = self.execute(params)
                duration = (time.time() - start) * 1000
                logger.info(f"工具 {self.name} 调用成功，耗时 {duration:.0f}ms")
                return ToolResult(success=True, data=data, duration_ms=duration)

            except TimeoutError:
                logger.warning(f"工具 {self.name} 超时（第 {attempt+1} 次尝试）")
                if attempt < self.max_retries:
                    time.sleep(1 * (attempt + 1))
                    continue
                duration = (time.time() - start) * 1000
                return ToolResult(success=False, error="timeout", duration_ms=duration)

            except Exception as e:
                logger.error(f"工具 {self.name} 调用失败: {e}")
                if attempt < self.max_retries and self._is_retryable(e):
                    time.sleep(1 * (attempt + 1))
                    continue
                duration = (time.time() - start) * 1000
                return ToolResult(success=False, error=str(e), duration_ms=duration)

        duration = (time.time() - start) * 1000
        return ToolResult(success=False, error="max_retries_exceeded", duration_ms=duration)

    def _is_retryable(self, error: Exception) -> bool:
        """判断错误是否可重试"""
        retryable_types = (ConnectionError, TimeoutError, APIError)
        return isinstance(error, retryable_types)

四、弹性扩缩容与成本优化

4.1 弹性扩缩容策略

Agent 服务的负载通常有明显的波峰波谷：白天工作时间负载高，夜间负载低。弹性扩缩容可以在保证服务质量的同时降低成本。

扩缩容的触发指标：

Kubernetes HPA（Horizontal Pod Autoscaler）配置示例：

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: agent-service-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: agent-service
  minReplicas: 3
  maxReplicas: 50
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
      - type: Percent
        value: 100
        periodSeconds: 60
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Percent
        value: 20
        periodSeconds: 120

扩缩容注意事项： – 扩容要快、缩容要慢：扩容时快速增加实例应对流量，缩容时缓慢减少避免抖动 – 设置合理的最小实例数，确保基础可用性 – 新实例启动需要时间（拉取镜像、加载模型、预热），扩容阈值要提前触发 – 缩容前要确保实例上的任务执行完成，使用优雅终止

4.2 成本构成与优化方向

AI Agent 系统的成本构成通常是：

LLM API 费用：60-80%（最大头）
GPU 推理服务器：10-20%（如果自建模型）
云服务器（CPU/内存）：5-10%
数据库与存储：2-5%
网络与其他：1-3%

成本优化的优先级应该按成本占比来：LLM API 费用占比最大，优化收益最高。

LLM API 成本优化方法：

1. 模型路由：简单任务用小模型，复杂任务用大模型
2. 缓存复用：相同请求直接返回缓存结果
3. 上下文精简：及时清理无用的上下文，减少输入 Token
4. 批量处理：多个小请求合并成一次调用
5. 限制最大输出：设置合理的 max_tokens，避免生成过长内容
6. 使用更便宜的模型：在效果可接受的前提下，优先用便宜的模型

上下文精简示例：

def trim_messages(messages: list, max_tokens: int = 8000) -> list:
    """精简消息历史，控制 Token 数量"""
    # 系统提示词始终保留
    system_msg = next((m for m in messages if m["role"] == "system"), None)

    # 按时间倒序保留消息，直到达到 Token 上限
    recent_messages = []
    total_tokens = 0

    for msg in reversed(messages):
        if msg["role"] == "system":
            continue
        msg_tokens = estimate_tokens(msg["content"])
        if total_tokens + msg_tokens > max_tokens:
            break
        recent_messages.insert(0, msg)
        total_tokens += msg_tokens

    # 重新组装
    result = []
    if system_msg:
        result.append(system_msg)
    if len(recent_messages) < len([m for m in messages if m["role"] != "system"]):
        result.append({
            "role": "system",
            "content": f"[注：为控制上下文长度，已省略 {len(messages) - len(recent_messages) - 1} 条历史消息]"
        })
    result.extend(recent_messages)
    return result

4.3 成本监控与预算告警

建立成本监控体系，实时跟踪费用支出，设置预算告警：

• 按服务维度：每个服务的成本占比和趋势
• 按模型维度：每个模型的调用次数、Token 消耗、费用
• 按用户维度：Top 用户的成本消耗，识别异常用户
• 按任务类型维度：哪类任务成本最高，是否有优化空间
• 时间趋势：日/周/月成本趋势，预测月度账单

预算告警设置： – 月度预算的 50% → 提醒通知 – 月度预算的 80% → 警告通知，排查异常 – 月度预算的 100% → 紧急通知，考虑限流或降级 – 单日费用超过日均 2 倍 → 异常告警

五、可观测性：日志、指标与链路追踪

5.1 可观测性三大支柱

生产环境的 Agent 系统必须具备完善的可观测性，否则出了问题只能靠猜。可观测性有三大支柱：

三者结合，才能从宏观到微观、从整体到细节地理解系统运行状态。

5.2 结构化日志

Agent 系统的日志必须是结构化的（JSON 格式），方便检索和分析。不要用 print() 输出非结构化日志。

结构化日志示例：

import logging
import json
from pythonjsonlogger import jsonlogger

logger = logging.getLogger("agent")

class AgentJsonFormatter(jsonlogger.JsonFormatter):
    def add_fields(self, log_record, record, message_dict):
        super().add_fields(log_record, record, message_dict)
        log_record["service"] = "agent-service"
        log_record["level"] = record.levelname

# 配置日志
handler = logging.StreamHandler()
handler.setFormatter(AgentJsonFormatter(
    "%(asctime)s %(levelname)s %(message)s %(session_id)s %(task_id)s %(tool)s %(duration_ms)s"
))
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# 使用示例
logger.info(
    "工具调用完成",
    extra={
        "session_id": "sess_123",
        "task_id": "task_456",
        "tool": "search_database",
        "duration_ms": 235,
        "success": True,
    }
)

Agent 系统的关键日志点： – 任务开始/结束（任务 ID、用户 ID、输入、输出、耗时） – 每一步决策（思考内容、选择的工具、工具参数） – 工具调用（工具名、参数、结果、耗时、是否重试） – LLM 调用（模型、输入 Token、输出 Token、耗时、费用） – 异常和错误（错误类型、错误信息、堆栈、上下文） – 状态变更（会话创建、任务状态变更、缓存命中/未命中）

日志级别规范： – DEBUG：详细的调试信息，生产环境默认关闭 – INFO：正常的业务流程记录（任务开始/结束、工具调用等） – WARNING：需要关注但不影响服务的异常（重试、降级、缓存未命中） – ERROR：影响单个请求的错误（工具调用失败、LLM 超时） – CRITICAL：影响整个服务的严重故障（数据库不可用、模型服务全部失败）

5.3 核心指标监控

使用 Prometheus 收集指标，Grafana 可视化。Agent 系统的核心指标包括：

业务指标： – agent_tasks_total：任务总数（按状态、类型、用户分标签） – agent_task_duration_seconds：任务执行耗时分布（Histogram） – agent_task_success_rate：任务成功率 – agent_active_sessions：当前活跃会话数

模型指标： – llm_requests_total：LLM 调用总数（按模型、状态分标签） – llm_tokens_total：Token 消耗总数（按模型、输入/输出分标签） – llm_duration_seconds：LLM 调用耗时分布 – llm_cost_dollars_total：LLM 费用累计

工具指标： – tool_calls_total：工具调用总数（按工具名、状态分标签） – tool_duration_seconds：工具调用耗时分布 – tool_retry_total：工具重试次数

系统指标： – CPU/内存/磁盘/网络利用率 – 请求队列长度 – 实例数（当前/期望） – 错误率（5xx、4xx）

Prometheus 指标定义示例：

from prometheus_client import Counter, Histogram, Gauge

# 任务计数器
task_counter = Counter(
    "agent_tasks_total",
    "Total number of agent tasks",
    ["status", "task_type"]
)

# 任务耗时直方图
task_duration = Histogram(
    "agent_task_duration_seconds",
    "Duration of agent tasks in seconds",
    ["task_type"],
    buckets=(1, 5, 10, 30, 60, 120, 300, 600)
)

# LLM Token 消耗
llm_tokens = Counter(
    "llm_tokens_total",
    "Total LLM tokens consumed",
    ["model", "token_type"]
)

# 当前活跃会话
active_sessions = Gauge(
    "agent_active_sessions",
    "Current active agent sessions"
)

# 使用示例
def execute_task(task_type: str):
    start = time.time()
    try:
        result = run_agent()
        task_counter.labels(status="success", task_type=task_type).inc()
        return result
    except Exception:
        task_counter.labels(status="failed", task_type=task_type).inc()
        raise
    finally:
        duration = time.time() - start
        task_duration.labels(task_type=task_type).observe(duration)

5.4 分布式链路追踪

一个 Agent 任务可能经过：API 网关 → Agent 服务 → LLM 服务 → 工具服务 → 数据库。链路追踪可以把这些环节串起来，清晰展示请求的完整路径和每个环节的耗时。

使用 OpenTelemetry 实现链路追踪：

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter

# 初始化追踪
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

jaeger_exporter = JaegerExporter(
    agent_host_name="jaeger",
    agent_port=6831,
)
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

# 在 Agent 执行中使用
def run_agent_task(task_input):
    with tracer.start_as_current_span("agent.task") as span:
        span.set_attribute("task.id", task_input["task_id"])
        span.set_attribute("user.id", task_input["user_id"])

        # LLM 调用
        with tracer.start_as_current_span("llm.call") as llm_span:
            llm_span.set_attribute("llm.model", "gpt-4")
            result = call_llm(messages)
            llm_span.set_attribute("llm.tokens.input", result["usage"]["prompt_tokens"])
            llm_span.set_attribute("llm.tokens.output", result["usage"]["completion_tokens"])

        # 工具调用
        with tracer.start_as_current_span(f"tool.{tool_name}") as tool_span:
            tool_span.set_attribute("tool.name", tool_name)
            tool_result = call_tool(tool_name, params)
            tool_span.set_attribute("tool.success", tool_result.success)

        return result

链路追踪的关键属性： – 每个 span 设置明确的名称和属性 – LLM 调用记录模型名、Token 数、耗时 – 工具调用记录工具名、参数摘要、成功状态 – 异常时记录错误信息和堆栈

六、故障处理与高可用设计

6.1 常见故障类型与应对

Agent 系统的故障类型比传统 Web 服务更多样，因为它依赖 LLM、工具、外部 API 等多个不稳定组件：

6.2 超时、重试与熔断

这三个模式是处理分布式系统故障的基础。

超时控制：每个外部调用都必须设置超时，避免无限等待：

# 各级超时设置（秒）
LLM_CALL_TIMEOUT = 30          # LLM 调用超时
TOOL_CALL_TIMEOUT = 15          # 工具调用超时
DB_QUERY_TIMEOUT = 10           # 数据库查询超时
TASK_EXECUTION_TIMEOUT = 600    # 整个任务超时（10分钟）

重试策略：对可重试的错误进行有限次数重试，使用指数退避：

import time
import random

def with_retry(func, max_retries=3, base_delay=1, max_delay=30):
    """带指数退避和抖动的重试"""
    for attempt in range(max_retries + 1):
        try:
            return func()
        except (TimeoutError, ConnectionError, APIError) as e:
            if attempt == max_retries:
                raise
            # 指数退避 + 随机抖动
            delay = min(base_delay * (2 ** attempt), max_delay)
            delay += random.uniform(0, delay * 0.1)
            time.sleep(delay)

熔断模式：当某个依赖持续失败时，暂时停止调用，快速失败，避免雪崩：

import time
from collections import deque

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_threshold = failure_threshold      # 连续失败阈值
        self.recovery_timeout = recovery_timeout        # 恢复超时（秒）
        self.failures = deque(maxlen=failure_threshold) # 最近失败记录
        self.state = "closed"  # closed(正常) / open(熔断) / half_open(半开)
        self.last_failure_time = None

    def call(self, func, *args, **kwargs):
        if self.state == "open":
            # 检查是否到了恢复时间
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half_open"
            else:
                raise CircuitBreakerOpen("熔断器已打开，快速失败")

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        if self.state == "half_open":
            self.state = "closed"
        self.failures.clear()

    def _on_failure(self):
        self.failures.append(time.time())
        self.last_failure_time = time.time()
        if len(self.failures) >= self.failure_threshold:
            self.state = "open"

6.3 高可用架构设计

高可用的目标是消除单点故障，确保任何一个组件故障时系统仍能正常运行。

关键设计：

1. 多实例部署：无状态服务至少部署 3 个实例，分布在不同可用区
2. 负载均衡：请求在多个实例间分发，实例故障时自动摘除
3. 主从复制：数据库、Redis 等有状态服务使用主从架构，主节点故障时自动切换
4. 多可用区部署：服务和数据分布在多个可用区，单个可用区故障不影响整体
5. 优雅终止：实例关闭前完成正在执行的任务，拒绝新请求，避免任务中断
6. 健康检查：定期检查实例健康状态，异常实例自动重启或摘除

优雅终止实现：

import signal
import time
import threading

class GracefulShutdown:
    def __init__(self):
        self.shutdown_requested = False
        self.active_tasks = 0
        self._lock = threading.Lock()

        # 注册信号处理
        signal.signal(signal.SIGTERM, self._handle_signal)
        signal.signal(signal.SIGINT, self._handle_signal)

    def _handle_signal(self, signum, frame):
        print(f"收到信号 {signum}，开始优雅关闭...")
        self.shutdown_requested = True

    def task_start(self):
        """任务开始时调用"""
        with self._lock:
            self.active_tasks += 1

    def task_end(self):
        """任务结束时调用"""
        with self._lock:
            self.active_tasks -= 1

    def wait_for_completion(self, timeout=60):
        """等待所有任务完成"""
        start = time.time()
        while self.active_tasks > 0:
            if time.time() - start > timeout:
                print(f"等待超时，仍有 {self.active_tasks} 个任务在执行，强制关闭")
                break
            time.sleep(1)
        print("优雅关闭完成")

# 使用示例
shutdown = GracefulShutdown()

@app.post("/agent/task")
async def handle_task(request):
    if shutdown.shutdown_requested:
        return JSONResponse(status_code=503, content={"error": "服务正在关闭"})

    shutdown.task_start()
    try:
        result = await execute_agent_task(request)
        return result
    finally:
        shutdown.task_end()

6.4 灾难恢复与备份

除了日常的高可用，还要考虑极端情况的灾难恢复：

• 数据备份：数据库定期全量备份 + 实时增量备份，备份数据存储在不同地域
• 配置版本化：所有配置文件纳入版本控制，可快速回滚到任意版本
• 基础设施即代码：用 Terraform/Ansible 管理基础设施，可在新环境快速重建
• 灾备演练：定期进行故障演练，验证恢复流程的有效性
• RTO/RPO 定义：明确恢复时间目标（RTO）和恢复点目标（RPO），指导备份策略

七、多环境管理与发布策略

7.1 环境划分

典型的多环境划分：

每个环境有独立的配置、数据库、模型服务，环境之间完全隔离。配置通过环境变量或配置中心注入，代码不硬编码环境相关信息。

7.2 配置管理

使用配置中心（如 Nacos、Apollo、Consul）管理各环境的配置，或者用环境变量 + .env 文件。

配置管理原则： – 代码与配置分离：配置不硬编码在代码中，通过环境变量或配置中心注入 – 敏感信息加密：API Key、密码等敏感信息加密存储，运行时解密 – 配置版本化：配置变更有记录、可回滚 – 环境隔离：不同环境的配置完全独立，禁止跨环境共享配置

配置加载示例：

import os
from dataclasses import dataclass
from typing import Optional

@dataclass
class Config:
    environment: str
    database_url: str
    redis_url: str
    llm_api_key: str
    llm_model: str
    log_level: str
    max_concurrent_tasks: int

    @classmethod
    def from_env(cls) -> "Config":
        return cls(
            environment=os.getenv("ENV", "dev"),
            database_url=os.getenv("DATABASE_URL", "postgresql://localhost:5432/agent_dev"),
            redis_url=os.getenv("REDIS_URL", "redis://localhost:6379/0"),
            llm_api_key=os.getenv("LLM_API_KEY", ""),
            llm_model=os.getenv("LLM_MODEL", "gpt-4"),
            log_level=os.getenv("LOG_LEVEL", "INFO"),
            max_concurrent_tasks=int(os.getenv("MAX_CONCURRENT_TASKS", "100")),
        )

config = Config.from_env()

7.3 发布策略

选择合适的发布策略，平衡发布速度和风险：

推荐使用灰度发布作为常规发布策略：

第1阶段：1% 流量 → 观察 30 分钟 → 无异常继续
第2阶段：5% 流量 → 观察 1 小时 → 无异常继续
第3阶段：20% 流量 → 观察 2 小时 → 无异常继续
第4阶段：50% 流量 → 观察 4 小时 → 无异常继续
第5阶段：100% 流量 → 全量发布

任何阶段出现异常（错误率升高、性能下降、用户投诉），立即回滚到旧版本。

7.4 数据库变更管理

Agent 系统的数据库变更需要特别小心，因为数据量大、服务不能停。使用数据库迁移工具（如 Alembic、Flyway）管理变更：

• 所有数据库变更通过迁移脚本执行，不手动改表
• 迁移脚本纳入版本控制，有明确的回滚脚本
• 大表变更使用在线 DDL 工具（如 pt-online-schema-change），避免锁表
• 破坏性变更（删表、删字段）分阶段执行：先兼容 → 再废弃 → 最后删除
• 发布前在预发环境验证迁移脚本

八、上线前检查清单

部署新版本前，逐项确认以下检查清单：

基础设施

• [ ] 容器镜像已构建并推送到镜像仓库，标签明确
• [ ] Kubernetes 配置（Deployment、Service、HPA、ConfigMap）已更新
• [ ] 资源请求和限制已设置（CPU、内存）
• [ ] 健康检查（就绪探针、存活探针）已配置
• [ ] 环境变量和密钥已正确配置
• [ ] 数据库迁移脚本已在预发环境验证通过

高可用

• [ ] 至少 3 个实例，分布在不同可用区
• [ ] 负载均衡配置正确，健康检查通过
• [ ] 优雅终止已实现，关闭信号处理正确
• [ ] 熔断、降级、重试策略已配置
• [ ] 超时设置合理（LLM、工具、数据库、整体任务）
• [ ] 回滚方案已准备，回滚操作已演练

性能与成本

• [ ] 压测已完成，满足预期 QPS 和延迟目标
• [ ] 弹性扩缩容配置正确，扩容阈值合理
• [ ] 模型路由策略已配置（简单任务用小模型）
• [ ] 缓存策略已启用，缓存命中率可接受
• [ ] 成本监控已配置，预算告警已设置
• [ ] 上下文精简逻辑已启用，避免无效 Token 消耗

可观测性

• [ ] 结构化日志已配置，关键日志点已覆盖
• [ ] Prometheus 指标已定义，Grafana 仪表盘已更新
• [ ] 链路追踪已接入，关键 span 已添加属性
• [ ] 告警规则已配置，告警通道已验证
• [ ] 核心指标基线已建立（完成率、延迟、成本、错误率）
• [ ] 日志级别设置合理（生产环境 INFO，非 DEBUG）

安全与合规

• [ ] 敏感信息（API Key、密码）未硬编码，通过 Secret 管理
• [ ] 数据库连接使用 SSL/TLS 加密
• [ ] API 鉴权和限流已配置
• [ ] 数据备份策略已验证，恢复流程已演练
• [ ] 操作日志完整可审计
• [ ] 个人数据处理符合隐私政策

发布执行

• [ ] 灰度发布计划已制定（1%→5%→20%→50%→100%）
• [ ] 发布时间选择低峰期
• [ ] 值班人员已通知，应急预案已确认
• [ ] 发布后验证清单已准备
• [ ] 用户公告已准备（如有重大变更）
• [ ] 相关团队已同步发布计划

结语

AI Agent 的部署与运维是一门实践性很强的工程学科。没有完美的架构，只有适合业务场景的架构。关键是在可用性、性能、成本、安全之间找到平衡点，并通过完善的可观测性和自动化运维，让系统能够持续稳定运行。

核心原则有三条：第一，无状态服务水平扩展，有状态数据独立托管，这是分布式系统的基本设计原则；第二，默认一切都会失败，为每个组件设计超时、重试、熔断、降级，让系统在故障中仍能提供核心服务；第三，可观测性先行，在系统上线前就把日志、指标、链路追踪建好，出了问题能快速定位根因。

至此，AI Agent 工程化实践系列已经覆盖了从提示词工程、Function Calling、任务规划、长期记忆、多智能体协作、评测质量保障到部署运维的完整技术链路。下一篇可以继续讨论 AI Agent 的安全与治理：从提示注入防护、权限最小化、数据脱敏、操作审计到合规治理，探讨如何在享受 Agent 强大能力的同时，把安全风险控制在可接受的范围内。