模块15: 部署与最佳实践

将你的智能体容器化可以确保它在开发、预发布和生产环境中一致运行。Docker容器将你的智能体代码、依赖项和配置打包成一个可部署的单元。

智能体服务的Dockerfile

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# Install dependencies first (cached layer)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY src/ ./src/
COPY prompts/ ./prompts/
COPY config/ ./config/

# Non-root user for security
RUN useradd --create-home appuser
USER appuser

# Health check endpoint
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
  CMD curl -f http://localhost:8000/health || exit 1

EXPOSE 8000
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

多智能体系统的Docker Compose

当你有多个智能体作为独立服务运行时（如模块11中讨论的），Docker Compose让你可以一起定义和运行它们。

# docker-compose.yml
version: "3.9"
services:
  orchestrator:
    build: ./orchestrator
    ports: ["8000:8000"]
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - REDIS_URL=redis://redis:6379
    depends_on: [redis]

  research-agent:
    build: ./agents/research
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}

  writer-agent:
    build: ./agents/writer
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}

  redis:
    image: redis:7-alpine
    volumes: ["redis-data:/data"]

volumes:
  redis-data:

生产中的智能体面临各种故障：API速率限制、网络超时、格式错误的LLM响应和工具执行错误。带有重试和熔断器的健壮错误处理使你的智能体保持弹性。

带抖动的指数退避

import time, random
import anthropic

def call_with_retry(func, max_retries: int = 3, base_delay: float = 1.0):
    """Call a function with exponential backoff and jitter on failure."""
    for attempt in range(max_retries + 1):
        try:
            return func()
        except anthropic.RateLimitError:
            if attempt == max_retries:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"Rate limited. Retrying in {delay:.1f}s (attempt {attempt + 1})")
            time.sleep(delay)
        except anthropic.APIStatusError as e:
            if e.status_code >= 500:  # Server errors are retryable
                if attempt == max_retries:
                    raise
                time.sleep(base_delay * (2 ** attempt))
            else:
                raise  # Client errors (4xx) are not retryable
        except anthropic.APIConnectionError:
            if attempt == max_retries:
                raise
            time.sleep(base_delay * (2 ** attempt))
    raise Exception("Max retries exceeded")

熔断器模式

如果API持续失败，停止请求并快速失败，而不是等待重试。熔断器在达到失败阈值后"打开"，在冷却期后"关闭"。

import time

class CircuitBreaker:
    """Prevent cascading failures with a circuit breaker."""

    def __init__(self, failure_threshold: int = 5, reset_timeout: float = 60.0):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failure_count = 0
        self.last_failure_time = 0
        self.state = "closed"  # closed = normal, open = failing fast

    def call(self, func):
        """Execute function through the circuit breaker."""
        if self.state == "open":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "half-open"  # Try one request
            else:
                raise Exception("Circuit breaker is OPEN - failing fast")

        try:
            result = func()
            if self.state == "half-open":
                self.state = "closed"
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
                print(f"Circuit breaker OPENED after {self.failure_count} failures")
            raise

# Usage
api_breaker = CircuitBreaker(failure_threshold=5, reset_timeout=60)
try:
    result = api_breaker.call(lambda: client.messages.create(...))
except Exception as e:
    # Serve cached response or graceful fallback
    result = fallback_response()

看不到的问题无法修复。生产中的智能体需要全面的日志记录、指标和追踪来诊断问题并跟踪长期性能。

结构化日志

import logging, json, time, uuid

class AgentLogger:
    """Structured logging for agent interactions."""

    def __init__(self, agent_name: str):
        self.logger = logging.getLogger(agent_name)
        self.logger.setLevel(logging.INFO)
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter('%(message)s'))
        self.logger.addHandler(handler)

    def log_llm_call(self, request_id: str, model: str, input_tokens: int,
                     output_tokens: int, latency_ms: float, success: bool):
        self.logger.info(json.dumps({
            "event": "llm_call",
            "request_id": request_id,
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "latency_ms": round(latency_ms, 2),
            "success": success,
            "cost_usd": self._estimate_cost(model, input_tokens, output_tokens),
            "timestamp": time.time(),
        }))

    def log_tool_call(self, request_id: str, tool_name: str,
                      latency_ms: float, success: bool, error: str = ""):
        self.logger.info(json.dumps({
            "event": "tool_call",
            "request_id": request_id,
            "tool": tool_name,
            "latency_ms": round(latency_ms, 2),
            "success": success,
            "error": error,
            "timestamp": time.time(),
        }))

    def _estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        prices = {
            "claude-sonnet-4-20250514": (3.0, 15.0),   # per 1M tokens
            "claude-haiku-4-20250514":  (0.25, 1.25),
        }
        input_price, output_price = prices.get(model, (3.0, 15.0))
        return (input_tokens * input_price + output_tokens * output_price) / 1_000_000

需要跟踪的关键指标

AI智能体主要是I/O密集型（等待LLM API响应），而非CPU密集型。这意味着扩展策略与传统的计算密集型应用不同。

使用FastAPI的水平扩展

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import asyncio, anthropic

app = FastAPI()
client = anthropic.AsyncAnthropic()

class AgentRequest(BaseModel):
    user_id: str
    message: str

class AgentResponse(BaseModel):
    response: str
    latency_ms: float

@app.post("/agent", response_model=AgentResponse)
async def handle_request(req: AgentRequest):
    """Handle agent requests asynchronously for high throughput."""
    import time
    start = time.time()

    response = await client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        system="You are a helpful assistant.",
        messages=[{"role": "user", "content": req.message}]
    )

    latency = (time.time() - start) * 1000
    return AgentResponse(
        response=response.content[0].text,
        latency_ms=latency
    )

@app.get("/health")
async def health():
    return {"status": "healthy"}

在生产中，LLM API成本可以迅速升级，特别是在多智能体系统中，每个用户请求会触发多个API调用。主动的成本管理至关重要。

成本优化策略

class CostOptimiser:
    """Route requests to the cheapest capable model."""

    def __init__(self):
        self.client = anthropic.Anthropic()
        self.model_tiers = {
            "simple":  "claude-haiku-4-20250514",    # Classification, extraction
            "medium":  "claude-sonnet-4-20250514",   # Analysis, writing
            "complex": "claude-opus-4-20250514",     # Complex reasoning
        }

    def classify_complexity(self, task: str) -> str:
        """Use a cheap model to classify task complexity."""
        response = self.client.messages.create(
            model="claude-haiku-4-20250514",
            max_tokens=10,
            system="Classify task complexity as: simple, medium, or complex. "
                   "One word only.",
            messages=[{"role": "user", "content": task}]
        )
        complexity = response.content[0].text.strip().lower()
        return complexity if complexity in self.model_tiers else "medium"

    def run(self, task: str) -> str:
        """Route to the appropriate model based on complexity."""
        complexity = self.classify_complexity(task)
        model = self.model_tiers[complexity]
        print(f"Routing to {model} (complexity: {complexity})")

        response = self.client.messages.create(
            model=model, max_tokens=2048,
            messages=[{"role": "user", "content": task}]
        )
        return response.content[0].text

关键的成本节省技术：

1. 模型路由 — 简单任务用便宜的模型，仅在需要时用昂贵的模型（节省5-50倍）
2. 提示缓存 — 缓存系统提示和常用上下文以减少输入token
3. 响应缓存 — 缓存相同或近似相同的查询以避免冗余API调用
4. Token预算 — 为每个任务适当设置 max_tokens；不要为是/否问题请求4096个token
5. 批处理 — 对非时间敏感的工作负载使用批量API（通常便宜50%）
6. 支出告警 — 在API密钥上设置硬性支出上限，并在接近限制时发出告警

在将智能体部署到生产之前，逐项检查这份全面的清单。每个项目都针对生产AI系统中常见的故障模式。

你已经完成了AI智能体系列的全部15个模块。你现在拥有了构建、测试和部署生产级AI智能体的基础知识和实践技能。以下是继续学习的建议路径：

• 通过关注 Anthropic、MCP 和 A2A 的官方文档，保持对快速发展的智能体生态系统的了解
• 加入社区论坛和Discord服务器，分享模式并向其他构建智能体的人学习
• 阅读关于智能体架构、工具使用和安全的研究论文——学术前沿发展很快