构建生产级 Agent：监控、安全与部署最佳实践

安全护栏：第一优先级

生产环境中，Agent 必须有严格的安全边界。核心防护措施包括：

• 输入校验：过滤恶意注入和超长输入
• 工具权限控制：限制 Agent 可访问的工具和资源范围
• 输出过滤：防止敏感信息泄露

from claude_agent_sdk import Agent, tool

@tool
def query_database(sql: str) -> str:
    """执行数据库查询（仅允许 SELECT）"""
    if not sql.strip().upper().startswith("SELECT"):
        return "错误：仅允许 SELECT 查询"
    # 执行查询...
    return "查询结果：..."

agent = Agent(
    name="数据查询助手",
    instructions="你只能执行只读数据库查询，禁止任何修改操作。",
    tools=[query_database],
    max_turns=10,  # 限制最大执行步数
)

Human-in-the-Loop：关键决策人工确认

对于高风险操作（如删除数据、发送通知），加入人工确认环节：

@tool
def delete_record(record_id: str) -> str:
    """删除记录（需要人工确认）"""
    confirmation = input(f"确认删除记录 {record_id}？(yes/no): ")
    if confirmation != "yes":
        return "操作已取消"
    # 执行删除...
    return f"记录 {record_id} 已删除"

监控与可观测性

生产级 Agent 必须具备完整的监控能力：

import logging

logger = logging.getLogger("agent")

@tool
def process_order(order_id: str) -> str:
    """处理订单"""
    logger.info(f"开始处理订单: {order_id}")
    try:
        result = "订单处理成功"
        logger.info(f"订单 {order_id} 处理完成")
        return result
    except Exception as e:
        logger.error(f"订单 {order_id} 处理失败: {e}")
        return f"处理失败：{str(e)}"

关键监控指标：每次运行的总步数、token 消耗量、工具调用成功率、端到端延迟。

成本控制策略

• 设置 max_turns 防止无限循环
• 简单子任务使用轻量模型，复杂推理使用高级模型
• 为每次调用设置 token 预算上限
• 缓存重复的工具调用结果

部署与扩展

生产部署的关键考虑：

• 无状态设计：Agent 执行不依赖本地状态，便于水平扩展
• 队列管理：使用消息队列（如 Redis/RabbitMQ）管理请求，避免突发流量压垮服务
• 速率限制：对用户和全局请求频率做限制
• 优雅降级：当 API 不可用时，返回预设的兜底响应

测试策略

• 工具单元测试：独立测试每个工具函数的输入输出
• 流程集成测试：模拟完整对话流，验证多步骤执行结果
• 边界测试：测试异常输入、工具超时、网络错误等场景

常见问题

Agent 在生产环境中如何处理 API 超时？

建议为每个工具调用设置超时时间，并实现重试机制（指数退避）。同时设置全局超时，防止单次 Agent 运行占用过长时间。对于关键任务，可以配置备用模型或降级策略。

如何防止 Agent 进入无限循环？

通过 max_turns 参数限制最大执行步数是最直接的方法。此外，可以在工具中添加计数器，检测重复调用模式。如果检测到循环行为，工具返回终止信号让 Agent 停止。

多个 Agent 实例如何协调？

在分布式环境中，使用消息队列管理任务分发，通过数据库或 Redis 共享状态。每个 Agent 实例应设计为无状态的，所有持久化数据存储在外部系统中。